redis流數(shù)據(jù)推送多用戶的方法是什么

/** * 關(guān)閉連接時(shí) */@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {  InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();  int port = ipSocket.getPort();  String host = ipSocket.getHostString();  String serverUrl = host + ":" + port;  String prev = redisTemplate.opsForValue().get(SOCKET_CONNECT_PREFIX + serverUrl);  Integer cur = Math.toIntExact(System.currentTimeMillis() / 1000);  if (null == prev || cur - Integer.parseInt(prev) > 20 * 60) {    redisTemplate.opsForValue().set(SOCKET_CONNECT_PREFIX + serverUrl, cur + "");    log.info("服務(wù)端斷開連接=====" + host + port + "20分鐘之后 重新連接");    final EventLoop eventLoop = ctx.channel().eventLoop();    Bootstrap bootstrap = defaultProcessHandler.getBootstrap(eventLoop);    eventLoop.schedule(        () -> defaultProcessHandler.doConnect(bootstrap, host + ":" + port, new AtomicInteger(0)),        20,        TimeUnit.MINUTES);  } else {    log.warn(serverUrl + "距離上次斷開連接不足20分鐘==" + (cur - Integer.parseInt(prev)) + "s");  }  super.channelInactive(ctx);}

serverBootstrap.childHandler(new ChannelInitializer<Channel>() {  @Override  protected void initChannel(Channel channel) {    //此方法每次客戶端連接都會(huì)調(diào)用，是為通道初始化的方法
    //獲得通道channel中的管道鏈（執(zhí)行鏈、handler鏈）    ChannelPipeline pipeline = channel.pipeline();    pipeline.addLast(new LineBasedFrameDecoder(Short.MAX_VALUE * 10));    pipeline.addLast(new StringDecoder());    pipeline.addLast(new StringHandler());
    log.info("success to initHandler!");  }});

ChannelFuture future = channels.get(callbackUrl.getUrl());if (null != future) {  try {    boolean result = sendMsg(future, pushDataStr);    this.redisTemplate.opsForValue().set(callbackUrl.getUrl(), offset.toString());    if (!result) {      log.error("this channel push failed {}", callbackUrl.getUrl());      returnVal = false;    }  } catch (Exception e) {    log.error("push exception", e);    returnVal = false;  }}return returnVal;