refactor(matching-engine): snapshot preload

Author: greensheng

src/main/java/com/gitbitex/Bootstrap.java

@@ -7,6 +7,7 @@
 import com.gitbitex.marketdata.manager.TradeManager;
 import com.gitbitex.marketdata.orderbook.OrderBookSnapshotManager;
 import com.gitbitex.marketdata.repository.CandleRepository;
+import com.gitbitex.matchingengine.MatchingEngineLoader;
 import com.gitbitex.matchingengine.MatchingEngineThread;
 import com.gitbitex.matchingengine.MessageSender;
 import com.gitbitex.matchingengine.command.Command;
@@ -40,6 +41,7 @@ public class Bootstrap {
     private final AppProperties appProperties;
     private final KafkaProperties kafkaProperties;
     private final EngineSnapshotManager engineSnapshotManager;
+    private final MatchingEngineLoader matchingEngineLoader;
     private final MessageSender messageSender;
     private final OrderBookSnapshotManager orderBookSnapshotManager;
     private final RedissonClient redissonClient;
@@ -61,7 +63,7 @@ private void startMatchingEngine(int nThreads) {
         for (int i = 0; i < nThreads; i++) {
             String groupId = "MatchingEngine";
             var consumer = getEngineCommandKafkaConsumer(groupId);
-            var thread = new MatchingEngineThread(consumer, engineSnapshotManager, messageSender, appProperties);
+            var thread = new MatchingEngineThread(consumer, matchingEngineLoader, appProperties);
             thread.setName(groupId + "-" + thread.getId());
             thread.setUncaughtExceptionHandler(getUncaughtExceptionHandler(() -> startMatchingEngine(1)));
             thread.start();

src/main/java/com/gitbitex/matchingengine/MatchingEngineLoader.java

@@ -0,0 +1,39 @@
+package com.gitbitex.matchingengine;
+
+import com.gitbitex.matchingengine.snapshot.EngineSnapshotManager;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Nullable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+public class MatchingEngineLoader {
+    private final EngineSnapshotManager engineSnapshotManager;
+    private final MessageSender messageSender;
+    @Getter
+    @Nullable
+    private volatile MatchingEngine preperedMatchingEngine;
+
+    public MatchingEngineLoader(EngineSnapshotManager engineSnapshotManager, MessageSender messageSender) {
+        this.engineSnapshotManager = engineSnapshotManager;
+        this.messageSender = messageSender;
+        startRefreshPreparingMatchingEnginePeriodically();
+    }
+
+    private void startRefreshPreparingMatchingEnginePeriodically() {
+        Executors.newScheduledThreadPool(1).scheduleWithFixedDelay(() -> {
+            try {
+                logger.info("reloading latest snapshot");
+                preperedMatchingEngine = new MatchingEngine(engineSnapshotManager, messageSender);
+                logger.info("done");
+            } catch (Exception e) {
+                logger.error("matching engine create error: {}", e.getMessage(), e);
+            }
+        }, 0, 1, TimeUnit.MINUTES);
+    }
+
+}

src/main/java/com/gitbitex/matchingengine/MatchingEngineThread.java

@@ -1,8 +1,7 @@
 package com.gitbitex.matchingengine;
 
 import com.gitbitex.AppProperties;
-import com.gitbitex.matchingengine.command.*;
-import com.gitbitex.matchingengine.snapshot.EngineSnapshotManager;
+import com.gitbitex.matchingengine.command.Command;
 import com.gitbitex.middleware.kafka.KafkaConsumerThread;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -17,16 +16,15 @@
 public class MatchingEngineThread extends KafkaConsumerThread<String, Command>
         implements ConsumerRebalanceListener {
     private final AppProperties appProperties;
-    private final EngineSnapshotManager engineSnapshotManager;
-    private final MessageSender messageSender;
+    private final MatchingEngineLoader matchingEngineLoader;
     private MatchingEngine matchingEngine;
 
-    public MatchingEngineThread(KafkaConsumer<String, Command> consumer, EngineSnapshotManager engineSnapshotManager, MessageSender messageSender,
+    public MatchingEngineThread(KafkaConsumer<String, Command> consumer, MatchingEngineLoader matchingEngineLoader,
                                 AppProperties appProperties) {
         super(consumer, logger);
         this.appProperties = appProperties;
-        this.engineSnapshotManager = engineSnapshotManager;
-        this.messageSender = messageSender;
+        this.matchingEngineLoader = matchingEngineLoader;
+
     }
 
     @Override
@@ -40,7 +38,10 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
     public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
         for (TopicPartition partition : partitions) {
             logger.info("partition assigned: {}", partition.toString());
-            matchingEngine = new MatchingEngine(engineSnapshotManager, messageSender);
+            matchingEngine = matchingEngineLoader.getPreperedMatchingEngine();
+            if (matchingEngine == null) {
+                throw new RuntimeException("no prepared matching engine");
+            }
             if (matchingEngine.getStartupCommandOffset() != null) {
                 logger.info("seek to offset: {}", matchingEngine.getStartupCommandOffset() + 1);
                 consumer.seek(partition, matchingEngine.getStartupCommandOffset() + 1);