[RIP-63] Queue Selection Strategy Optimization

client/src/main/java/org/apache/rocketmq/client/ClientConfig.java

@@ -38,6 +38,8 @@ public class ClientConfig {
     public static final String SOCKS_PROXY_CONFIG = "com.rocketmq.socks.proxy.config";
     public static final String DECODE_READ_BODY = "com.rocketmq.read.body";
     public static final String DECODE_DECOMPRESS_BODY = "com.rocketmq.decompress.body";
+    public static final String SEND_LATENCY_ENABLE = "com.rocketmq.sendLatencyEnable";
+    public static final String START_DETECTOR_ENABLE = "com.rocketmq.startDetectorEnable";
     public static final String HEART_BEAT_V2 = "com.rocketmq.heartbeat.v2";
     private String namesrvAddr = NameServerAddressUtils.getNameServerAddresses();
     private String clientIP = NetworkUtil.getLocalAddress();
@@ -72,6 +74,8 @@ public class ClientConfig {
     private String socksProxyConfig = System.getProperty(SOCKS_PROXY_CONFIG, "{}");
 
     private int mqClientApiTimeout = 3 * 1000;
+    private int detectTimeout = 200;
+    private int detectInterval = 2 * 1000;
 
     private LanguageCode language = LanguageCode.JAVA;
 
@@ -81,6 +85,15 @@ public class ClientConfig {
      */
     protected boolean enableStreamRequestType = false;
 
+    /**
+     * Enable the fault tolerance mechanism of the client sending process.
+     * DO NOT OPEN when ORDER messages are required.
+     * Turning on will interfere with the queue selection functionality,
+     * possibly conflicting with the order message.
+     */
+    private boolean sendLatencyEnable = Boolean.parseBoolean(System.getProperty(SEND_LATENCY_ENABLE, "false"));
+    private boolean startDetectorEnable = Boolean.parseBoolean(System.getProperty(START_DETECTOR_ENABLE, "false"));
+
     public String buildMQClientId() {
         StringBuilder sb = new StringBuilder();
         sb.append(this.getClientIP());
@@ -186,6 +199,10 @@ public void resetClientConfig(final ClientConfig cc) {
         this.decodeDecompressBody = cc.decodeDecompressBody;
         this.enableStreamRequestType = cc.enableStreamRequestType;
         this.useHeartbeatV2 = cc.useHeartbeatV2;
+        this.startDetectorEnable = cc.startDetectorEnable;
+        this.sendLatencyEnable = cc.sendLatencyEnable;
+        this.detectInterval = cc.detectInterval;
+        this.detectTimeout = cc.detectTimeout;
     }
 
     public ClientConfig cloneClientConfig() {
@@ -210,6 +227,10 @@ public ClientConfig cloneClientConfig() {
         cc.decodeDecompressBody = decodeDecompressBody;
         cc.enableStreamRequestType = enableStreamRequestType;
         cc.useHeartbeatV2 = useHeartbeatV2;
+        cc.startDetectorEnable = startDetectorEnable;
+        cc.sendLatencyEnable = sendLatencyEnable;
+        cc.detectInterval = detectInterval;
+        cc.detectTimeout = detectTimeout;
         return cc;
     }
 
@@ -381,6 +402,38 @@ public void setEnableStreamRequestType(boolean enableStreamRequestType) {
         this.enableStreamRequestType = enableStreamRequestType;
     }
 
+    public boolean isSendLatencyEnable() {
+        return sendLatencyEnable;
+    }
+
+    public void setSendLatencyEnable(boolean sendLatencyEnable) {
+        this.sendLatencyEnable = sendLatencyEnable;
+    }
+
+    public boolean isStartDetectorEnable() {
+        return startDetectorEnable;
+    }
+
+    public void setStartDetectorEnable(boolean startDetectorEnable) {
+        this.startDetectorEnable = startDetectorEnable;
+    }
+
+    public int getDetectTimeout() {
+        return this.detectTimeout;
+    }
+
+    public void setDetectTimeout(int detectTimeout) {
+        this.detectTimeout = detectTimeout;
+    }
+
+    public int getDetectInterval() {
+        return this.detectInterval;
+    }
+
+    public void setDetectInterval(int detectInterval) {
+        this.detectInterval = detectInterval;
+    }
+
     public boolean isUseHeartbeatV2() {
         return useHeartbeatV2;
     }
@@ -403,6 +456,7 @@ public String toString() {
             + ", socksProxyConfig=" + socksProxyConfig + ", language=" + language.name()
             + ", namespace=" + namespace + ", mqClientApiTimeout=" + mqClientApiTimeout
             + ", decodeReadBody=" + decodeReadBody + ", decodeDecompressBody=" + decodeDecompressBody
+            + ", sendLatencyEnable=" + sendLatencyEnable + ", startDetectorEnable=" + startDetectorEnable
             + ", enableStreamRequestType=" + enableStreamRequestType + ", useHeartbeatV2=" + useHeartbeatV2 + "]";
     }
 }

client/src/main/java/org/apache/rocketmq/client/common/ThreadLocalIndex.java

@@ -33,6 +33,14 @@ public int incrementAndGet() {
         return index & POSITIVE_MASK;
     }
 
+    public void reset() {
+        int index = Math.abs(random.nextInt(Integer.MAX_VALUE));
+        if (index < 0) {
+            index = 0;
+        }
+        this.threadLocalIndex.set(index);
+    }
+
     @Override
     public String toString() {
         return "ThreadLocalIndex{" +

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -666,7 +666,7 @@ public void operationComplete(ResponseFuture responseFuture) {
                         } catch (Throwable e) {
                         }
 
-                        producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false);
+                        producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false, true);
                         return;
                     }
 
@@ -684,14 +684,14 @@ public void operationComplete(ResponseFuture responseFuture) {
                             } catch (Throwable e) {
                             }
 
-                            producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false);
+                            producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false, true);
                         } catch (Exception e) {
-                            producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
+                            producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true, true);
                             onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
                                 retryTimesWhenSendFailed, times, e, context, false, producer);
                         }
                     } else {
-                        producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
+                        producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true, true);
                         if (!responseFuture.isSendRequestOK()) {
                             MQClientException ex = new MQClientException("send request failed", responseFuture.getCause());
                             onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
@@ -711,7 +711,7 @@ public void operationComplete(ResponseFuture responseFuture) {
             });
         } catch (Exception ex) {
             long cost = System.currentTimeMillis() - beginStartTime;
-            producer.updateFaultItem(brokerName, cost, true);
+            producer.updateFaultItem(brokerName, cost, true, false);
             onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
                 retryTimesWhenSendFailed, times, ex, context, true, producer);
         }
@@ -735,7 +735,7 @@ private void onExceptionImpl(final String brokerName,
         if (needRetry && tmp <= timesTotal) {
             String retryBrokerName = brokerName;//by default, it will send to the same broker
             if (topicPublishInfo != null) { //select one message queue accordingly, in order to determine which broker to send
-                MessageQueue mqChosen = producer.selectOneMessageQueue(topicPublishInfo, brokerName);
+                MessageQueue mqChosen = producer.selectOneMessageQueue(topicPublishInfo, brokerName, false);
                 retryBrokerName = instance.getBrokerNameFromMessageQueue(mqChosen);
             }
             String addr = instance.findBrokerAddressInPublish(retryBrokerName);

client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Lock;
@@ -125,6 +126,12 @@ public class MQClientInstance {
     private final Set<String/* Broker address */> brokerSupportV2HeartbeatSet = new HashSet();
     private final ConcurrentMap<String, Integer> brokerAddrHeartbeatFingerprintTable = new ConcurrentHashMap();
     private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "MQClientFactoryScheduledThread"));
+    private final ScheduledExecutorService fetchRemoteConfigExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+        @Override
+        public Thread newThread(Runnable r) {
+            return new Thread(r, "MQClientFactoryFetchRemoteConfigScheduledThread");
+        }
+    });
     private final PullMessageService pullMessageService;
     private final RebalanceService rebalanceService;
     private final DefaultMQProducer defaultMQProducer;