mirror of
https://github.com/apache/rocketmq.git
synced 2026-09-24 16:04:00 +08:00
ltamber
e38da5590f
* add compaction delete policy
* wrapper message ext encoder
* fix
* cleanup policy
* cleanup policy
* wrapper message ext encoder
* Revert "cleanup policy"
This reverts commit da76a4820b.
* cleanup policy
* topic compaction
* compaction recovery and user document
* use cleanup policy
* fix
53 lines
1.7 KiB
Markdown
53 lines
1.7 KiB
Markdown
# Compaction Topic
|
|
|
|
## use example
|
|
### create compaction topic
|
|
```shell
|
|
$ bin/mqadmin updateTopic -w 8 -r 8 -a +delete.policy=COMPACTION -n localhost:9876 -t ctopic -c DefaultCluster
|
|
create topic to 127.0.0.1:10911 success.
|
|
TopicConfig [topicName=ctopic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={+delete.policy=COMPACTION}]
|
|
```
|
|
|
|
### produce message
|
|
the same with ordinary message
|
|
```java
|
|
DefaultMQProducer producer = new DefaultMQProducer("CompactionTestGroup");
|
|
producer.setNamesrvAddr("localhost:9876");
|
|
producer.start();
|
|
|
|
Message msg = new Message(topic, "tags", "keys", "bodys"getBytes(StandardCharsets.UTF_8));
|
|
SendResult sendResult = producer.send(msg);
|
|
|
|
System.out.printf("%s%n", sendResult);
|
|
```
|
|
### consume message
|
|
the message offset remains unchanged after compaction. If the consumer specified offset does not exist, return the most recent message after the offset.
|
|
|
|
In the compaction scenario, most consumption was started from the beginning of the queue.
|
|
```java
|
|
DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("compactionTestGroup");
|
|
consumer.setNamesrvAddr("localhost:9876");
|
|
consumer.setPullThreadNums(4);
|
|
consumer.start();
|
|
|
|
Collection<MessageQueue> messageQueueList = consumer.fetchMessageQueues("ctopic");
|
|
consumer.assign(messageQueueList);
|
|
messageQueueList.forEach(mq -> {
|
|
try {
|
|
consumer.seekToBegin(mq);
|
|
} catch (MQClientException e) {
|
|
e.printStackTrace();
|
|
}
|
|
});
|
|
|
|
Map<String, byte[]> kvStore = Maps.newHashMap();
|
|
while (true) {
|
|
List<MessageExt> msgList = consumer.poll(1000);
|
|
if (msgList != null) {
|
|
msgList.forEach(msg -> kvStore.put(msg.getKeys(), msg.getBody()));
|
|
}
|
|
}
|
|
|
|
//use the kvStore
|
|
```
|