add redis client name

ctripcorp · Nov 6, 2024 · bdca901 · bdca901
1 parent 841869b
commit bdca901
Show file tree

Hide file tree

Showing 23 changed files with 346 additions and 34 deletions.
diff --git a/core/src/main/java/com/ctrip/xpipe/netty/commands/AbstractNettyRequestResponseCommand.java b/core/src/main/java/com/ctrip/xpipe/netty/commands/AbstractNettyRequestResponseCommand.java
@@ -56,16 +56,21 @@ protected void doSendRequest(final NettyClient nettyClient, ByteBuf byteBuf) {
 		}
 
 		if(getCommandTimeoutMilli() > 0 && scheduled != null){
+			int commandTimeoutMilli = getCommandTimeoutMilli();
+			if (nettyClient instanceof RedisNettyClient && !((RedisNettyClient) nettyClient).getDoAfterConnectedOver()) {
+				commandTimeoutMilli += ((RedisNettyClient) nettyClient).getAfterConnectCommandTimeoutMill();
+			}
 
-			getLogger().debug("[doSendRequest][schedule timeout]{}, {}", this, getCommandTimeoutMilli());
+			getLogger().debug("[doSendRequest][schedule timeout]{}, {}", this, commandTimeoutMilli);
+			int finalCommandTimeoutMilli = commandTimeoutMilli;
 			timeoutFuture = scheduled.schedule(new AbstractExceptionLogTask() {
 
 				@Override
 				public void doRun() {
 					getLogger().info("[{}][run][timeout]{}", AbstractNettyRequestResponseCommand.this, nettyClient);
-					future().setFailure(new CommandTimeoutException("timeout " +  + getCommandTimeoutMilli()));
+					future().setFailure(new CommandTimeoutException("timeout " + finalCommandTimeoutMilli));
 				}
-			}, getCommandTimeoutMilli(), TimeUnit.MILLISECONDS);
+			}, commandTimeoutMilli, TimeUnit.MILLISECONDS);
 
 			future().addListener(new CommandFutureListener<V>() {
 

diff --git a/core/src/main/java/com/ctrip/xpipe/netty/commands/AsyncNettyClient.java b/core/src/main/java/com/ctrip/xpipe/netty/commands/AsyncNettyClient.java
@@ -9,6 +9,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.atomic.AtomicReference;
+
 
 /**
  * @author chen.zhu
@@ -21,6 +23,8 @@ public class AsyncNettyClient extends DefaultNettyClient {
 
     private ChannelFuture future;
 
+    protected final AtomicReference<Boolean> doAfterConnectedOver = new AtomicReference<>(false);
+
     public AsyncNettyClient(ChannelFuture future, Endpoint endpoint) {
         super(future.channel());
         this.future = future;
@@ -35,6 +39,8 @@ public void operationComplete(ChannelFuture future) {
                     } else {
                         desc.set(ChannelUtil.getDesc(future.channel()));
                     }
+                    doAfterConnected();
+                    doAfterConnectedOver.set(true);
                 } else {
                     logger.info("[async][connect-fail] endpint: {}", endpoint, future.cause());
                 }
@@ -93,4 +99,16 @@ public String toString() {
         return super.toString();
     }
 
+    protected void doAfterConnected() {
+
+    }
+
+    protected boolean getDoAfterConnectedOver() {
+        return doAfterConnectedOver.get();
+    }
+
+    protected int getAfterConnectCommandTimeoutMill() {
+        return 0;
+    }
+
 }
diff --git a/core/src/main/java/com/ctrip/xpipe/netty/commands/NettyClientHandler.java b/core/src/main/java/com/ctrip/xpipe/netty/commands/NettyClientHandler.java
@@ -46,7 +46,11 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
 
 			@Override
 			public void read(Channel channel, ByteBuf byteBuf) throws ByteBufReadActionException {
-				nettyClient.handleResponse(channel, byteBuf);
+				try {
+					nettyClient.handleResponse(channel, byteBuf);
+				} catch (Exception e) {
+					logger.error("[NettyClientHandler][channelRead]", e);
+				}
 			}
 		});
 

diff --git a/core/src/main/java/com/ctrip/xpipe/netty/commands/RedisNettyClient.java b/core/src/main/java/com/ctrip/xpipe/netty/commands/RedisNettyClient.java
@@ -0,0 +1,9 @@
+package com.ctrip.xpipe.netty.commands;
+
+public interface RedisNettyClient {
+
+    boolean getDoAfterConnectedOver();
+
+    int getAfterConnectCommandTimeoutMill();
+
+}
diff --git a/core/src/main/java/com/ctrip/xpipe/pool/XpipeNettyClientPool.java b/core/src/main/java/com/ctrip/xpipe/pool/XpipeNettyClientPool.java
@@ -61,7 +61,7 @@ public NettyClient borrowObject() throws BorrowObjectException {
 		try {
 			return objectPool.borrowObject();
 		} catch (Exception e) {
-			logger.error("[borrowObject]" + factory,  e);
+			logger.error("[borrowObject] NumIdle:{}, NumActive:{}" + factory, objectPool.getNumIdle(), objectPool.getNumActive(), e);
 			throw new BorrowObjectException("borrow " + factory, e);
 		}
 	}

diff --git a/redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/CheckerConfig.java b/redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/config/CheckerConfig.java
@@ -112,4 +112,6 @@ public interface CheckerConfig {
 
     int getMarkInstanceMaxDelayMilli();
 
+    boolean getDoAfterNettyClientConnected();
+
 }
diff --git a/...checker/src/main/java/com/ctrip/xpipe/redis/checker/config/impl/DataCenterConfigBean.java b/...checker/src/main/java/com/ctrip/xpipe/redis/checker/config/impl/DataCenterConfigBean.java
@@ -37,6 +37,8 @@ public class DataCenterConfigBean extends AbstractConfigBean {
 
     public static final String KEY_HTTP_ACCEPT_ENCODING = "http.accept.encoding";
 
+    private static final String KEY_NETTY_CLIENT_DO_AFTER_CONNECTED = "netty.client.do.after.connected";
+
     private AtomicReference<String> zkConnection = new AtomicReference<>();
     private AtomicReference<String> zkNameSpace = new AtomicReference<>();
 
@@ -95,4 +97,8 @@ public String getHttpAcceptEncoding() {
         return getProperty(KEY_HTTP_ACCEPT_ENCODING, null);
     }
 
+    public boolean getDoAfterNettyClientConnected() {
+        return getBooleanProperty(KEY_NETTY_CLIENT_DO_AFTER_CONNECTED, true);
+    }
+
 }
diff --git a/...a/com/ctrip/xpipe/redis/checker/healthcheck/actions/redismaster/MasterOverOneMonitor.java b/...a/com/ctrip/xpipe/redis/checker/healthcheck/actions/redismaster/MasterOverOneMonitor.java
@@ -33,7 +33,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Collectors;
 
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.REDIS_KEYED_NETTY_CLIENT_POOL;
 import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.SCHEDULED_EXECUTOR;
 
 /**
@@ -49,7 +49,7 @@ public class MasterOverOneMonitor implements RedisMasterActionListener, OneWaySu
     @Autowired
     private AlertManager alertManager;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = REDIS_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     @Resource(name = SCHEDULED_EXECUTOR)

diff --git a/...e/redis/checker/healthcheck/actions/sentinel/collector/DefaultSentinelHelloCollector.java b/...e/redis/checker/healthcheck/actions/sentinel/collector/DefaultSentinelHelloCollector.java
@@ -46,7 +46,7 @@
 import java.util.stream.Collectors;
 
 import static com.ctrip.xpipe.redis.checker.healthcheck.actions.sentinel.SentinelHelloCheckAction.LOG_TITLE;
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.SENTINEL_KEYED_NETTY_CLIENT_POOL;
 import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.THREAD_POOL_TIME_OUT;
 
 /**
@@ -77,7 +77,7 @@ public class DefaultSentinelHelloCollector implements SentinelHelloCollector {
     @Autowired
     private PersistenceCache persistenceCache;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = SENTINEL_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     private SentinelLeakyBucket leakyBucket;

diff --git a/...r/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/impl/ProxyConnectedChecker.java b/...r/src/main/java/com/ctrip/xpipe/redis/checker/healthcheck/impl/ProxyConnectedChecker.java
@@ -19,7 +19,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ScheduledExecutorService;
 
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.PROXY_KEYED_NETTY_CLIENT_POOL;
 import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.SCHEDULED_EXECUTOR;
 
 @Component
@@ -28,7 +28,7 @@ public class ProxyConnectedChecker implements ProxyChecker {
     @Autowired
     private CheckerConfig checkerConfig;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = PROXY_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     @Resource(name = SCHEDULED_EXECUTOR)

diff --git a/redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/resource/Resource.java b/redis/redis-checker/src/main/java/com/ctrip/xpipe/redis/checker/resource/Resource.java
@@ -8,11 +8,15 @@ public class Resource {
 
     public static final String REDIS_COMMAND_EXECUTOR = "redisCommandExecutor";
 
-    public static final String KEYED_NETTY_CLIENT_POOL = "keyedClientPool";
+    public static final String REDIS_KEYED_NETTY_CLIENT_POOL = "redisKeyedClientPool";
 
-    public static final String REDIS_SESSION_NETTY_CLIENT_POOL = "redisSessionClientPool";
+    public static final String SENTINEL_KEYED_NETTY_CLIENT_POOL = "sentinelKeyedClientPool";
+
+    public static final String KEEPER_KEYED_NETTY_CLIENT_POOL = "keeperKeyedClientPool";
 
-    public static final String MIGRATE_KEEPER_CLIENT_POOL = "migrateKeeperClientPool";
+    public static final String PROXY_KEYED_NETTY_CLIENT_POOL = "proxyKeyedClientPool";
+
+    public static final String REDIS_SESSION_NETTY_CLIENT_POOL = "redisSessionClientPool";
 
     public static final String PING_DELAY_INFO_EXECUTORS = "pingDelayInfoExecutors";
 

diff --git a/...s-checker/src/test/java/com/ctrip/xpipe/redis/checker/AbstractCheckerIntegrationTest.java b/...s-checker/src/test/java/com/ctrip/xpipe/redis/checker/AbstractCheckerIntegrationTest.java
@@ -5,6 +5,7 @@
 import com.ctrip.xpipe.concurrent.DefaultExecutorFactory;
 import com.ctrip.xpipe.lifecycle.LifecycleHelper;
 import com.ctrip.xpipe.netty.commands.NettyKeyedPoolClientFactory;
+import com.ctrip.xpipe.redis.core.client.NettyRedisPoolClientFactory;
 import com.ctrip.xpipe.pool.XpipeNettyClientKeyedObjectPool;
 import com.ctrip.xpipe.redis.checker.cluster.GroupCheckerLeaderElector;
 import com.ctrip.xpipe.redis.checker.config.CheckerConfig;
@@ -148,17 +149,33 @@ public ScheduledExecutorService getRedisCommandExecutor() {
             );
         }
 
-        @Bean(name = KEYED_NETTY_CLIENT_POOL)
+        @Bean(name = REDIS_KEYED_NETTY_CLIENT_POOL)
         public XpipeNettyClientKeyedObjectPool getReqResNettyClientPool() throws Exception {
-            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(8));
+            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getRedisPoolClientFactory(8));
             LifecycleHelper.initializeIfPossible(keyedObjectPool);
             LifecycleHelper.startIfPossible(keyedObjectPool);
             return keyedObjectPool;
         }
 
         @Bean(name = REDIS_SESSION_NETTY_CLIENT_POOL)
         public XpipeNettyClientKeyedObjectPool getRedisSessionNettyClientPool() throws Exception {
-            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(12));
+            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getRedisPoolClientFactory(12));
+            LifecycleHelper.initializeIfPossible(keyedObjectPool);
+            LifecycleHelper.startIfPossible(keyedObjectPool);
+            return keyedObjectPool;
+        }
+
+        @Bean(name = SENTINEL_KEYED_NETTY_CLIENT_POOL)
+        public XpipeNettyClientKeyedObjectPool getSentinelReqResNettyClientPool() throws Exception {
+            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(8));
+            LifecycleHelper.initializeIfPossible(keyedObjectPool);
+            LifecycleHelper.startIfPossible(keyedObjectPool);
+            return keyedObjectPool;
+        }
+
+        @Bean(name = PROXY_KEYED_NETTY_CLIENT_POOL)
+        public XpipeNettyClientKeyedObjectPool getProxyReqResNettyClientPool() throws Exception {
+            XpipeNettyClientKeyedObjectPool keyedObjectPool = new XpipeNettyClientKeyedObjectPool(getKeyedPoolClientFactory(8));
             LifecycleHelper.initializeIfPossible(keyedObjectPool);
             LifecycleHelper.startIfPossible(keyedObjectPool);
             return keyedObjectPool;
@@ -195,6 +212,10 @@ public ScheduledExecutorService getHelloCheckScheduled() {
         private NettyKeyedPoolClientFactory getKeyedPoolClientFactory(int eventLoopThreads) {
             return new NettyKeyedPoolClientFactory(eventLoopThreads);
         }
+
+        private NettyKeyedPoolClientFactory getRedisPoolClientFactory(int eventLoopThreads) {
+            return new NettyRedisPoolClientFactory(eventLoopThreads, "xpipe", () -> true);
+        }
 
         @Bean
         public FoundationService foundationService() {

diff --git a/redis/redis-checker/src/test/java/com/ctrip/xpipe/redis/checker/TestConfig.java b/redis/redis-checker/src/test/java/com/ctrip/xpipe/redis/checker/TestConfig.java
@@ -315,4 +315,9 @@ public int getMarkInstanceMaxDelayMilli() {
         return 2000;
     }
 
+    @Override
+    public boolean getDoAfterNettyClientConnected() {
+        return true;
+    }
+
 }
diff --git a/...console/src/main/java/com/ctrip/xpipe/redis/console/config/impl/DefaultConsoleConfig.java b/...console/src/main/java/com/ctrip/xpipe/redis/console/config/impl/DefaultConsoleConfig.java
@@ -587,6 +587,11 @@ public String getHttpAcceptEncoding() {
         return dataCenterConfigBean.getHttpAcceptEncoding();
     }
 
+    @Override
+    public boolean getDoAfterNettyClientConnected() {
+        return dataCenterConfigBean.getDoAfterNettyClientConnected();
+    }
+
     @Override
     public void addListener(ConfigKeyListener listener) {
         this.listenersSet.add(listener);

diff --git a/...in/java/com/ctrip/xpipe/redis/console/proxy/impl/DefaultProxyMonitorCollectorManager.java b/...in/java/com/ctrip/xpipe/redis/console/proxy/impl/DefaultProxyMonitorCollectorManager.java
@@ -35,7 +35,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.PROXY_KEYED_NETTY_CLIENT_POOL;
 import static com.ctrip.xpipe.spring.AbstractSpringConfigContext.SCHEDULED_EXECUTOR;
 
 @Component
@@ -49,7 +49,7 @@ public class DefaultProxyMonitorCollectorManager extends AbstractStartStoppable
     @Resource(name = SCHEDULED_EXECUTOR)
     private ScheduledExecutorService scheduled;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = PROXY_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     @Autowired

diff --git a/...dis-console/src/main/java/com/ctrip/xpipe/redis/console/redis/DefaultSentinelManager.java b/...dis-console/src/main/java/com/ctrip/xpipe/redis/console/redis/DefaultSentinelManager.java
@@ -36,8 +36,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-import static com.ctrip.xpipe.redis.checker.resource.Resource.KEYED_NETTY_CLIENT_POOL;
-import static com.ctrip.xpipe.redis.checker.resource.Resource.REDIS_COMMAND_EXECUTOR;
+import static com.ctrip.xpipe.redis.checker.resource.Resource.*;
 
 /**
  * @author chen.zhu
@@ -53,7 +52,7 @@ public class DefaultSentinelManager implements SentinelManager, ShardEventHandle
 
     private static final int LONG_SENTINEL_COMMAND_TIMEOUT = 2000;
 
-    @Resource(name = KEYED_NETTY_CLIENT_POOL)
+    @Resource(name = SENTINEL_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedClientPool;
 
     @Resource(name = REDIS_COMMAND_EXECUTOR)

diff --git a/...src/main/java/com/ctrip/xpipe/redis/console/service/model/impl/ShardModelServiceImpl.java b/...src/main/java/com/ctrip/xpipe/redis/console/service/model/impl/ShardModelServiceImpl.java
@@ -69,7 +69,7 @@ public class ShardModelServiceImpl implements ShardModelService{
     @Resource(name = REDIS_COMMAND_EXECUTOR)
     private ScheduledExecutorService scheduled;
 
-    @Resource(name = MIGRATE_KEEPER_CLIENT_POOL)
+    @Resource(name = KEEPER_KEYED_NETTY_CLIENT_POOL)
     private XpipeNettyClientKeyedObjectPool keyedObjectPool;
 
     private RetryCommandFactory<Object> retryCommandFactory;
Original file line number	Diff line number	Diff line change
Expand Up		@@ -112,4 +112,6 @@ public interface CheckerConfig {

		int getMarkInstanceMaxDelayMilli();

		boolean getDoAfterNettyClientConnected();

		}