[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;