Optimize delayed message initialization logic, (#31)

speed up rop loading.
streamnative · Jul 8, 2021 · 200b91a · 200b91a
1 parent 1082884
commit 200b91a
Showing 1 changed file with 26 additions and 22 deletions.
diff --git a/...src/main/java/org/streamnative/pulsar/handlers/rocketmq/inner/ScheduleMessageService.java b/...src/main/java/org/streamnative/pulsar/handlers/rocketmq/inner/ScheduleMessageService.java
@@ -192,34 +192,15 @@ class DeliverDelayedMessageTimerTask extends TimerTask {
         private static final int MAX_BATCH_SIZE = 500;
         private final PulsarService pulsarService;
         private final int delayLevel;
-        private final Consumer<byte[]> delayedConsumer;
         private final RopEntryFormatter formatter = new RopEntryFormatter();
         private final SystemTimer timeoutTimer;
 
+        private Consumer<byte[]> delayedConsumer = null;
+
         public DeliverDelayedMessageTimerTask(int delayLevel) {
             this.delayLevel = delayLevel;
-            String delayTopic = getDelayedTopicName(delayLevel);
             this.pulsarService = ScheduleMessageService.this.pulsarBroker.pulsar();
             this.timeoutTimer = SystemTimer.builder().executorName("DeliverDelayedMessageTimeWheelExecutor").build();
-            try {
-                log.warn("Before create delayed consumer, the client config value: [{}]",
-                        ((PulsarClientImpl) this.pulsarService.getClient()).getConfiguration());
-                this.delayedConsumer = this.pulsarService.getClient()
-                        .newConsumer()
-                        .receiverQueueSize(MAX_FETCH_MESSAGE_NUM)
-                        .subscriptionMode(SubscriptionMode.Durable)
-                        .subscriptionType(SubscriptionType.Failover)
-                        .subscriptionName(getDelayedTopicConsumerName(delayLevel))
-                        .topic(delayTopic)
-                        .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
-                        .subscribe();
-
-                log.warn("The client config value: [{}]",
-                        ((PulsarClientImpl) this.pulsarService.getClient()).getConfiguration());
-            } catch (Exception e) {
-                log.error("create delayed topic[delayLevel={}] consumer error.", delayLevel, e);
-                throw new RuntimeException("Create delayed topic error");
-            }
         }
 
         public void close() {
@@ -236,7 +217,9 @@ public void advanceClock(long timeoutMs) {
         @Override
         public void run() {
             try {
-                Preconditions.checkNotNull(this.delayedConsumer);
+                if (delayedConsumer == null) {
+                    createConsumer();
+                }
                 int i = 0;
                 while (i++ < MAX_BATCH_SIZE && timeoutTimer.size() < MAX_BATCH_SIZE
                         && ScheduleMessageService.this.isStarted()) {
@@ -332,6 +315,27 @@ public void run() {
             }
         }
 
+        private void createConsumer() {
+            try {
+                log.debug("Before create delayed consumer, the client config value: [{}]",
+                        ((PulsarClientImpl) this.pulsarService.getClient()).getConfiguration());
+                this.delayedConsumer = this.pulsarService.getClient()
+                        .newConsumer()
+                        .receiverQueueSize(MAX_FETCH_MESSAGE_NUM)
+                        .subscriptionMode(SubscriptionMode.Durable)
+                        .subscriptionType(SubscriptionType.Failover)
+                        .subscriptionName(getDelayedTopicConsumerName(delayLevel))
+                        .topic(getDelayedTopicName(delayLevel))
+                        .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+                        .subscribe();
+                log.debug("The client config value: [{}]",
+                        ((PulsarClientImpl) this.pulsarService.getClient()).getConfiguration());
+            } catch (Exception e) {
+                log.error("Create delayed topic[delayLevel={}] consumer error.", delayLevel, e);
+                throw new RuntimeException("Create delay consumer error");
+            }
+        }
+
         private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {
             MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
             msgInner.setBody(msgExt.getBody());