devbox
AllinToyou
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

redission收发命令流程分析_redisson 发送原始报文

作者：AllinToyou | 2024-06-13 03:46:33

赞

踩

redisson 发送原始报文

一、示例，我们从最简单的GET命令开始。

RBucket<Object> t = redissonClient.getBucket("syncTradeUid_idOff");
int idOff = (int)t.get();

二、springboot的Redission自动配置

  1. @Order(value = 4001)
  2. @ConditionalOnProperty("redisson.password")
  3. @Configuration
  4. @EnableConfigurationProperties({RedissonProperties.class})
  5. public class RedissonAutoConfiguration {
  6.  
  7.     public RedissonAutoConfiguration() {
  8.        System.out.println("==========================redis 初始化成功=======================");
  9.     }
  10.  
  11.     @Autowired
  12.     private RedissonProperties redissonProperties;
  13.  
  14.     @Bean(name = "redissonClient")
  15.     @ConditionalOnProperty(name="redisson.address")
  16.     RedissonClient redissonSingle() {
  17.         Config config = new Config();
  18.         config.setCodec(new FastJsonCodec());
  19.         SingleServerConfig serverConfig = config.useSingleServer()
  20.                 .setAddress(redissonProperties.getAddress())
  21.                 .setTimeout(redissonProperties.getTimeout())
  22.                 .setConnectionPoolSize(redissonProperties.getConnectionPoolSize())
  23.                 .setConnectionMinimumIdleSize(redissonProperties.getConnectionMinimumIdleSize());
  24.  
  25.         if(!StrUtil.isEmpty(redissonProperties.getPassword())) {
  26.             serverConfig.setPassword(redissonProperties.getPassword());
  27.         }
  28.  
  29.         return Redisson.create(config);
  30.     }
  31.  
  32.     /**
  33.      * 哨兵模式自动装配
  34.      * @return
  35.      */
  36.     @Bean(name = "redissonClient")
  37.     @ConditionalOnProperty(name="redisson.masterName")
  38.     RedissonClient redissonSentinel() {
  39.         Config config = new Config();
  40.         config.setCodec(new FastJsonCodec());
  41.         SentinelServersConfig serverConfig = config.useSentinelServers().addSentinelAddress(redissonProperties.getSentinelAddresses())
  42.                 .setMasterName(redissonProperties.getMasterName())
  43.                 .setTimeout(redissonProperties.getTimeout())
  44.                 .setMasterConnectionPoolSize(redissonProperties.getMasterConnectionPoolSize())
  45.                 .setSlaveConnectionPoolSize(redissonProperties.getSlaveConnectionPoolSize())
  46.                 .setReadMode(ReadMode.SLAVE);
  47.  
  48.         if(!StrUtil.isEmpty(redissonProperties.getPassword())) {
  49.             serverConfig.setPassword(redissonProperties.getPassword());
  50.         }
  51.         return Redisson.create(config);
  52.     }
  53.  
  54.  
  55. }

application.properites

  1. #单机
  2. redisson.address = redis://127.0.0.1:6379
  3. redisson.password =
  4.  
  5. #哨兵
  6. #redisson.masterName=BF-20190319DBXF
  7. #redisson.schema=redis://
  8. #redisson.sentinelAddresses=redis://127.0.0.1:26379,redis://127.0.0.1:26479,redis://127.0.0.1:26579
  9. #redisson.password=

三、REDISSION自动配置初始化流程

1.从Redisson.create(config)创建redission对象开始。Redission继承于

RedissonClient

2. 创建连接管理器对象

  1. org.redisson.config.ConfigSupport
  2.   
  3. public static ConnectionManager createConnectionManager(Config configCopy) {
  4.         UUID id = UUID.randomUUID();
  5.         
  6.         if (configCopy.getMasterSlaveServersConfig() != null) {
  7.             validate(configCopy.getMasterSlaveServersConfig());
  8.             return new MasterSlaveConnectionManager(configCopy.getMasterSlaveServersConfig(), configCopy, id);
  9.         } else if (configCopy.getSingleServerConfig() != null) {
  10.             validate(configCopy.getSingleServerConfig());
  11.             return new SingleConnectionManager(configCopy.getSingleServerConfig(), configCopy, id);
  12.         } else if (configCopy.getSentinelServersConfig() != null) {
  13.             validate(configCopy.getSentinelServersConfig());
  14.             return new SentinelConnectionManager(configCopy.getSentinelServersConfig(), configCopy, id);
  15.         } else if (configCopy.getClusterServersConfig() != null) {
  16.             validate(configCopy.getClusterServersConfig());
  17.             return new ClusterConnectionManager(configCopy.getClusterServersConfig(), configCopy, id);
  18.         } else if (configCopy.getReplicatedServersConfig() != null) {
  19.             validate(configCopy.getReplicatedServersConfig());
  20.             return new ReplicatedConnectionManager(configCopy.getReplicatedServersConfig(), configCopy, id);
  21.         } else if (configCopy.getConnectionManager() != null) {
  22.             return configCopy.getConnectionManager();
  23.         }else {
  24.             throw new IllegalArgumentException("server(s) address(es) not defined!");
  25.         }
  26.     }

3.先看下MasterSlaveEntry->setupMasterEntry，这里会创建RedisClient,以及连接REDIS服务器。

  1. org.redisson.connection.MasterSlaveEntry    
  2.  
  3. public RFuture<RedisClient> setupMasterEntry(RedisURI address) {
  4.         RedisClient client = connectionManager.createClient(NodeType.MASTER, address, sslHostname);
  5.         return setupMasterEntry(client);
  6.     }

4.创建RedisClient,这里面单机也是使用主从管理器，即是只有主没有从。统一起来。

  1. org.redisson.connection.MasterSlaveConnectionManager
  2.     
  3. @Override
  4.     public RedisClient createClient(NodeType type, RedisURI address, int timeout, int commandTimeout, String sslHostname) {
  5.         RedisClientConfig redisConfig = createRedisConfig(type, address, timeout, commandTimeout, sslHostname);
  6.         return RedisClient.create(redisConfig);
  7.     }

5.在RedisClient会创建NEETY的bootstrap,channel,handler.

  1. org.redisson.client.RedisClient
  2.     
  3. private Bootstrap createBootstrap(RedisClientConfig config, Type type) {
  4.         Bootstrap bootstrap = new Bootstrap()
  5.                         .resolver(config.getResolverGroup())
  6.                         .channel(config.getSocketChannelClass())
  7.                         .group(config.getGroup());
  9.         bootstrap.handler(new RedisChannelInitializer(bootstrap, config, this, channels, type));
  10.         bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout());
  11.         bootstrap.option(ChannelOption.SO_KEEPALIVE, config.isKeepAlive());
  12.         bootstrap.option(ChannelOption.TCP_NODELAY, config.isTcpNoDelay());
  13.         config.getNettyHook().afterBoostrapInitialization(bootstrap);
  14.         return bootstrap;
  15.     }

 6.我们再看下RedisChannelInitializer，有添加哪些inBounder,outBounder

  1. org.redisson.client.handler.RedisChannelInitializer   
  2.  @Override
  3.     protected void initChannel(Channel ch) throws Exception {
  4.         initSsl(config, ch);
  5.         
  6.         if (type == Type.PLAIN) {
  7.             ch.pipeline().addLast(new RedisConnectionHandler(redisClient));
  8.         } else {
  9.             ch.pipeline().addLast(new RedisPubSubConnectionHandler(redisClient));
  10.         }
  11.  
  12.         ch.pipeline().addLast(
  13.             connectionWatchdog,
  14.             CommandEncoder.INSTANCE,
  15.             CommandBatchEncoder.INSTANCE,
  16.             new CommandsQueue());
  17.  
  18.         if (pingConnectionHandler != null) {
  19.             ch.pipeline().addLast(pingConnectionHandler);
  20.         }
  21.         
  22.         if (type == Type.PLAIN) {
  23.             ch.pipeline().addLast(new CommandDecoder(config.getAddress().getScheme()));
  24.         } else {
  25.             ch.pipeline().addLast(new CommandPubSubDecoder(config));
  26.         }
  27.  
  28.         ch.pipeline().addLast(new ErrorsLoggingHandler());
  29.  
  30.         config.getNettyHook().afterChannelInitialization(ch);
  31.     }

 

 7.创建好RedisClient后，开始连接REDIS服务器。这里首先异步解析地址，解析成功后，在添加到写连接池时会创建和添加连接，在创建连接时会去连接REDIS服务器。

  1.  org.redisson.connection. MasterSlaveEntry
  2.  private RFuture<RedisClient> setupMasterEntry(RedisClient client) {
  3.         RPromise<RedisClient> result = new RedissonPromise<RedisClient>();
  4.         result.onComplete((res, e) -> {
  5.             if (e != null) {
  6.                 client.shutdownAsync();
  7.             }
  8.         });
  9.         RFuture<InetSocketAddress> addrFuture = client.resolveAddr();
  10.         addrFuture.onComplete((res, e) -> {
  11.             if (e != null) {
  12.                 result.tryFailure(e);
  13.                 return;
  14.             }
  15.             
  16.             masterEntry = new ClientConnectionsEntry(
  17.                     client, 
  18.                     config.getMasterConnectionMinimumIdleSize(), 
  19.                     config.getMasterConnectionPoolSize(),
  20.                     config.getSubscriptionConnectionMinimumIdleSize(),
  21.                     config.getSubscriptionConnectionPoolSize(), 
  22.                     connectionManager,
  23.                     NodeType.MASTER);
  24.     
  25.             int counter = 1;
  26.             if (config.getSubscriptionMode() == SubscriptionMode.MASTER) {
  27.                 counter++;
  28.             }
  29.             
  30.             CountableListener<RedisClient> listener = new CountableListener<>(result, client, counter);
  31.             RFuture<Void> writeFuture = writeConnectionPool.add(masterEntry);
  32.             writeFuture.onComplete(listener);
  33.             
  34.         });
  35.         
  36.         return result;
  37.     }

8.查看连接REDIS服务器过程

  1. org.redisson.connection.pool.ConnectionPool.    
  2. private void initConnections(ClientConnectionsEntry entry, RPromise<Void> initPromise, boolean checkFreezed) {
  3.         int minimumIdleSize = getMinimumIdleSize(entry);
  4.  
  5.         if (minimumIdleSize == 0 || (checkFreezed && entry.isFreezed())) {
  6.             initPromise.trySuccess(null);
  7.             return;
  8.         }
  9.  
  10.         AtomicInteger initializedConnections = new AtomicInteger(minimumIdleSize);
  11.         int startAmount = Math.min(10, minimumIdleSize);
  12.         AtomicInteger requests = new AtomicInteger(startAmount);
  13.         for (int i = 0; i < startAmount; i++) {
  14.             createConnection(checkFreezed, requests, entry, initPromise, minimumIdleSize, initializedConnections);
  15.         }
  16.     }

在这里可以看到会初始化10个客户端连接到连接池。

9.从连接池去申请创建连接

  1. ConnectionPool 
  2. private void createConnection(boolean checkFreezed, AtomicInteger requests, ClientConnectionsEntry entry, RPromise<Void> initPromise,
  3.             int minimumIdleSize, AtomicInteger initializedConnections) {
  4.       
  5.         acquireConnection(entry, new Runnable() {
  6.             
  7.             @Override
  8.             public void run() {
  9.                 RPromise<T> promise = new RedissonPromise<T>();
  10.                 createConnection(entry, promise);
  11.                 promise.onComplete((conn, e) -> {
  12.  
  13.  
  14.                            
  15.                 });
  16.             }
  17.         });
  18.  
  19.     }

 10.最终创建连接是在RedisClient.connectAsync这个异步连接方法中。

  1.   public RFuture<RedisConnection> connectAsync() {
  2.         final RPromise<RedisConnection> f = new RedissonPromise<RedisConnection>();
  3.         
  4.         RFuture<InetSocketAddress> addrFuture = resolveAddr();
  5.         addrFuture.onComplete((res, e) -> {
  6.             if (e != null) {
  7.                 f.tryFailure(e);
  8.                 return;
  9.             }
  10.             
  11.             ChannelFuture channelFuture = bootstrap.connect(res);
  12.         });
  13.         
  14.         return f;
  15.     }

11.在连接成功后，RedisConnectionHandler.channelRegistered方法中创建连接对象。

  1.     @Override
  2.     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  3.         if (connection == null) {
  4.             connection = createConnection(ctx);
  5.         }
  6.         super.channelRegistered(ctx);
  7.     }

12.在这里对channel赋值，保存。这里每个channel里面会有一个RedisConnection的属性。

  1. RedisConnection    
  2. public void updateChannel(Channel channel) {
  3.         this.channel = channel;
  4.         channel.attr(CONNECTION).set(this);
  5.     }

13.在连接成功后发送PING心跳命令

  1.  BaseConnectionHandler
  2. @Override
  3.     public void channelActive(final ChannelHandlerContext ctx) {
  4.         List<RFuture<Object>> futures = new ArrayList<RFuture<Object>>();
  6.         RedisClientConfig config = redisClient.getConfig();
  7.         if (config.getPassword() != null) {
  8.             RFuture<Object> future;
  9.             if (config.getUsername() != null) {
  10.                 future = connection.async(RedisCommands.AUTH, config.getUsername(), config.getPassword());
  11.             } else {
  12.                 future = connection.async(RedisCommands.AUTH, config.getPassword());
  13.             }
  14.             futures.add(future);
  15.         }futures.add(future);
  16.         }
  17.         if (config.getPingConnectionInterval() > 0) {
  18.             RFuture<Object> future = connection.async(RedisCommands.PING);
  19.             futures.add(future);
  20.         }
  21.         
  22.   
  23.         
  24.         final AtomicBoolean retry = new AtomicBoolean();
  25.         final AtomicInteger commandsCounter = new AtomicInteger(futures.size());
  26.         for (RFuture<Object> future : futures) {
  27.             future.onComplete((res, e) -> {
  28.                 if (e != null) {
  29.                     if (e instanceof RedisLoadingException) {
  30.                         if (retry.compareAndSet(false, true)) {
  31.                             ctx.executor().schedule(() -> {
  32.                                 channelActive(ctx);
  33.                             }, 1, TimeUnit.SECONDS);
  34.                         }
  35.                         return;
  36.                     }
  37.                     connection.closeAsync();
  38.                     connectionPromise.tryFailure(e);
  39.                     return;
  40.                 }
  41.                 if (commandsCounter.decrementAndGet() == 0) {
  42.                     ctx.fireChannelActive();
  43.                     connectionPromise.trySuccess(connection);
  44.                 }
  45.             });
  46.         }
  47.     }

四、发送命令过程。

1.首先从RedissionBucket的set方法

,这里面的 commandExecutor来源于connectionManager中的命令执行器。

2.然后进行入到RedisExecutor中的execute方法，去异步执行命令。这里首先从连接池获取连接，然后在异步连接成功后，发送命令。

  1.   public void execute() {
  2.         codec = getCodec(codec);
  3.         
  4.         RFuture<RedisConnection> connectionFuture = getConnection();
  5.  
  6.         connectionFuture.onComplete((connection, e) -> {
  7.          
  8.             sendCommand(attemptPromise, connection);
  9.  
  10.             writeFuture.addListener(new ChannelFutureListener() {
  11.                 @Override
  12.                 public void operationComplete(ChannelFuture future) throws Exception {
  13.                     checkWriteFuture(writeFuture, attemptPromise, connection);
  14.                 }
  15.             });
  16.  
  17.             releaseConnection(attemptPromise, connectionFuture);
  18.         });
  19.  
  20.     }

3.获取连接是从连接池中获取。根据读写模式从连接管理器中选择可用连接返回。

  1. RedisExecutor    
  2. protected RFuture<RedisConnection> getConnection() {
  3.         if (readOnlyMode) {
  4.             connectionFuture = connectionManager.connectionReadOp(source, command);
  5.         } else {
  6.             connectionFuture = connectionManager.connectionWriteOp(source, command);
  7.         }
  8.         return connectionFuture;
  9.     }

 3.接着调用RedisConnection的send向channel写入数据。

  1. RedisConnection    
  2. public <T, R> ChannelFuture send(CommandData<T, R> data) {
  3.         return channel.writeAndFlush(data);
  4.     }

4.netty的inBoundHandler中有一个CommandsQueue，为一个命令同步队列，同一时刻一个连接只有一个命令在执行，执行完后，再执行下一个命令。

  1.  org.redisson.client.handler.CommandsQueue
  2. private void sendData(Channel ch) {
  3.         QueueCommandHolder command = queue.peek();
  4.         if (command != null && command.trySend()) {
  5.             QueueCommand data = command.getCommand();
  6.             List<CommandData<Object, Object>> pubSubOps = data.getPubSubOperations();
  7.             if (!pubSubOps.isEmpty()) {
  8.                 for (CommandData<Object, Object> cd : pubSubOps) {
  9.                     for (Object channel : cd.getParams()) {
  10.                         ch.pipeline().get(CommandPubSubDecoder.class).addPubSubCommand((ChannelName) channel, cd);
  11.                     }
  12.                 }
  13.             } else {
  14.                 ch.attr(CURRENT_COMMAND).set(data);
  15.             }
  16.  
  17.             command.getChannelPromise().addListener(listener);
  18.             ch.writeAndFlush(data, command.getChannelPromise());
  19.         }
  20.     }

 

 

五、接收数据回调过程。

1.接收inhandler， 在收到数据后，从attr中的current_command属性中取出数据。

  1.  CommandDecoder
  2. @Override
  3.     protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
  4.         QueueCommand data = ctx.channel().attr(CommandsQueue.CURRENT_COMMAND).get();
  5.  
  6.         if (state() == null) {
  7.             state(new State());
  8.         }
  9.         
  10.         if (data == null) {
  11.             while (in.writerIndex() > in.readerIndex()) {
  12.                 int endIndex = skipCommand(in);
  13.  
  14.                 try {
  15.                     decode(ctx, in, data);
  16.                 } catch (Exception e) {
  17.                     in.readerIndex(endIndex);
  18.                     throw e;
  19.                 }
  20.             }
  21.         } else {
  22.             int endIndex = 0;
  23.             if (!(data instanceof CommandsData)) {
  24.                 endIndex = skipCommand(in);
  25.             }
  26.             
  27.             try {
  28.                 decode(ctx, in, data);
  29.             } catch (Exception e) {
  30.                 if (!(data instanceof CommandsData)) {
  31.                     in.readerIndex(endIndex);
  32.                 }
  33.                 throw e;
  34.             }
  35.         }
  36.     }

 

2.根据相应的PROMISE设置回调数据。

  1.    CommandDecoder
  2.  protected void completeResponse(CommandData<Object, Object> data, Object result) {
  3.         if (data != null) {
  4.             data.getPromise().trySuccess(result);
  5.         }
  6.     }

3.在等待异步PROMISE结果。

  1.     CommandAsyncService
  2. @Override
  3.     public <V> V get(RFuture<V> future) {
  4.         try {
  5.             future.await();
  6.         } catch (InterruptedException e) {
  7.             Thread.currentThread().interrupt();
  8.         }
  9.         if (future.isSuccess()) {
  10.             return future.getNow();
  11.         }
  12.  
  13.         throw convertException(future);
  14.     }

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/AllinToyou/article/detail/710746
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号