optimize get channel remoteAddr

Author: 1782935682

broker/src/main/java/org/apache/rocketmq/broker/client/ProducerManager.java

@@ -85,15 +85,15 @@ public ProducerTableInfo getProducerTable() {
                 if (map.containsKey(group)) {
                     map.get(group).add(new ProducerInfo(
                         clientChannelInfo.getClientId(),
-                        clientChannelInfo.getChannel().remoteAddress().toString(),
+                        RemotingHelper.parseChannelRemoteAddr(clientChannelInfo.getChannel()),
                         clientChannelInfo.getLanguage(),
                         clientChannelInfo.getVersion(),
                         clientChannelInfo.getLastUpdateTimestamp()
                     ));
                 } else {
                     map.put(group, new ArrayList<>(Collections.singleton(new ProducerInfo(
                         clientChannelInfo.getClientId(),
-                        clientChannelInfo.getChannel().remoteAddress().toString(),
+                        RemotingHelper.parseChannelRemoteAddr(clientChannelInfo.getChannel()),
                         clientChannelInfo.getLanguage(),
                         clientChannelInfo.getVersion(),
                         clientChannelInfo.getLastUpdateTimestamp()

broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java

@@ -19,8 +19,8 @@
 import com.alibaba.fastjson.JSON;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
-import java.util.BitSet;
 import java.nio.charset.StandardCharsets;
+import java.util.BitSet;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.metrics.PopMetricsManager;
 import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
@@ -407,7 +407,7 @@ protected void ackOrderly(String topic, String consumeGroup, int qId, long ackOf
             if (nextOffset > -1) {
                 if (!this.brokerController.getConsumerOffsetManager().hasOffsetReset(topic, consumeGroup, qId)) {
                     this.brokerController.getConsumerOffsetManager().commitOffset(
-                        channel.remoteAddress().toString(), consumeGroup, topic, qId, nextOffset);
+                        RemotingHelper.parseChannelRemoteAddr(channel), consumeGroup, topic, qId, nextOffset);
                 }
                 if (!this.brokerController.getConsumerOrderInfoManager().checkBlock(null, topic, consumeGroup, qId, invisibleTime)) {
                     this.brokerController.getPopMessageProcessor().notifyMessageArriving(topic, qId, consumeGroup);
@@ -451,7 +451,7 @@ protected void ackOrderlyNew(String topic, String consumeGroup, int qId, long ac
             long nextOffset = consumerOrderInfoManager.commitAndNext(topic, consumeGroup, qId, ackOffset, popTime);
             if (brokerController.getBrokerConfig().isPopConsumerKVServiceLog()) {
                 POP_LOGGER.info("PopConsumerService ack orderly, time={}, topicId={}, groupId={}, queueId={}, " +
-                        "offset={}, next={}", popTime, topic, consumeGroup, qId, ackOffset, nextOffset);
+                    "offset={}, next={}", popTime, topic, consumeGroup, qId, ackOffset, nextOffset);
             }
 
             if (nextOffset > -1L) {

broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java

@@ -756,7 +756,7 @@ private CompletableFuture<Long> popMsgFromQueue(String topic, String attemptId,
                     // because offset in PopBuffer is not committed.
                     POP_LOGGER.warn("Pop initial offset, because store is no correct, {}, {}->{}",
                         lockKey, atomicOffset.get(), result.getNextBeginOffset());
-                    this.brokerController.getConsumerOffsetManager().commitOffset(channel.remoteAddress().toString(), requestHeader.getConsumerGroup(), topic,
+                    this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel), requestHeader.getConsumerGroup(), topic,
                         queueId, result.getNextBeginOffset());
                     atomicOffset.set(result.getNextBeginOffset());
                     return this.brokerController.getMessageStore().getMessageAsync(requestHeader.getConsumerGroup(), topic, queueId, atomicOffset.get(),
@@ -793,7 +793,7 @@ private CompletableFuture<Long> popMsgFromQueue(String topic, String attemptId,
                             requestHeader.getConsumerGroup(),
                             queueId, popTime, requestHeader.getInvisibleTime(), result.getMessageQueueOffset(),
                             orderCountInfo);
-                        this.brokerController.getConsumerOffsetManager().commitOffset(channel.remoteAddress().toString(),
+                        this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel),
                             requestHeader.getConsumerGroup(), topic, queueId, finalOffset);
                     } else {
                         if (!appendCheckPoint(requestHeader, topic, reviveQid, queueId, finalOffset, result, popTime, this.brokerController.getBrokerConfig().getBrokerName())) {
@@ -809,7 +809,7 @@ private CompletableFuture<Long> popMsgFromQueue(String topic, String attemptId,
                     || GetMessageStatus.NO_MATCHED_LOGIC_QUEUE.equals(result.getStatus()))
                     && result.getNextBeginOffset() > -1) {
                     if (isOrder) {
-                        this.brokerController.getConsumerOffsetManager().commitOffset(channel.remoteAddress().toString(), requestHeader.getConsumerGroup(), topic,
+                        this.brokerController.getConsumerOffsetManager().commitOffset(RemotingHelper.parseChannelRemoteAddr(channel), requestHeader.getConsumerGroup(), topic,
                             queueId, result.getNextBeginOffset());
                     } else {
                         popBufferMergeService.addCkMock(requestHeader.getConsumerGroup(), topic, queueId, finalOffset,

remoting/src/main/java/org/apache/rocketmq/remoting/common/RemotingHelper.java

@@ -361,7 +361,7 @@ public static CompletableFuture<Void> convertChannelFutureToCompletableFuture(Ch
             if (future.isSuccess()) {
                 completableFuture.complete(null);
             } else {
-                completableFuture.completeExceptionally(new RemotingConnectException(channelFuture.channel().remoteAddress().toString(), future.cause()));
+                completableFuture.completeExceptionally(new RemotingConnectException(RemotingHelper.parseChannelRemoteAddr(channelFuture.channel()), future.cause()));
             }
         });
         return completableFuture;

remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java

@@ -502,9 +502,9 @@ public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingComma
             return invokeImpl(channel, request, timeoutMillis).thenApply(ResponseFuture::getResponseCommand)
                 .get(timeoutMillis, TimeUnit.MILLISECONDS);
         } catch (ExecutionException e) {
-            throw new RemotingSendRequestException(channel.remoteAddress().toString(), e.getCause());
+            throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e.getCause());
         } catch (TimeoutException e) {
-            throw new RemotingTimeoutException(channel.remoteAddress().toString(), timeoutMillis, e.getCause());
+            throw new RemotingTimeoutException(RemotingHelper.parseChannelRemoteAddr(channel), timeoutMillis, e.getCause());
         }
     }
 

remoting/src/main/java/org/apache/rocketmq/remoting/netty/ResponseFuture.java

@@ -21,6 +21,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
 import org.apache.rocketmq.remoting.common.SemaphoreReleaseOnlyOnce;
 import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
@@ -45,12 +46,13 @@ public class ResponseFuture {
     private volatile boolean interrupted = false;
 
     public ResponseFuture(Channel channel, int opaque, long timeoutMillis, InvokeCallback invokeCallback,
-                          SemaphoreReleaseOnlyOnce once) {
+        SemaphoreReleaseOnlyOnce once) {
         this(channel, opaque, null, timeoutMillis, invokeCallback, once);
     }
 
-    public ResponseFuture(Channel channel, int opaque, RemotingCommand request, long timeoutMillis, InvokeCallback invokeCallback,
-                          SemaphoreReleaseOnlyOnce once) {
+    public ResponseFuture(Channel channel, int opaque, RemotingCommand request, long timeoutMillis,
+        InvokeCallback invokeCallback,
+        SemaphoreReleaseOnlyOnce once) {
         this.channel = channel;
         this.opaque = opaque;
         this.request = request;
@@ -67,9 +69,9 @@ public void executeInvokeCallback() {
                     invokeCallback.operationSucceed(response);
                 } else {
                     if (!isSendRequestOK()) {
-                        invokeCallback.operationFail(new RemotingSendRequestException(channel.remoteAddress().toString(), getCause()));
+                        invokeCallback.operationFail(new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), getCause()));
                     } else if (isTimeout()) {
-                        invokeCallback.operationFail(new RemotingTimeoutException(channel.remoteAddress().toString(), getTimeoutMillis(), getCause()));
+                        invokeCallback.operationFail(new RemotingTimeoutException(RemotingHelper.parseChannelRemoteAddr(channel), getTimeoutMillis(), getCause()));
                     } else {
                         invokeCallback.operationFail(new RemotingException(getRequestCommand().toString(), getCause()));
                     }