mirror of https://github.com/apache/rocketmq.git synced 2026-09-19 10:54:54 +08:00

Files

T

ltamber e38da5590f [ISSUE #3799 ]main compaction process (#5351 )

* 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

2022-10-24 16:21:39 +08:00

1.7 KiB

Raw Blame History

Compaction Topic

use example

create compaction topic

$ 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

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.

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

1.7 KiB Raw Blame History