Introducing new features.心跳完善

@@ -24,8 +24,8 @@ public class CIMClientHandleInitializer extends ChannelInitializer<Channel> {
     @Override
     @Override
     protected void initChannel(Channel ch) throws Exception {
     protected void initChannel(Channel ch) throws Exception {
         ch.pipeline()
         ch.pipeline()
-                //45 秒没发送消息 将IdleStateHandler 添加到 ChannelPipeline 中
-                .addLast(new IdleStateHandler(0, 45, 0))
+                //30 秒没发送消息 将IdleStateHandler 添加到 ChannelPipeline 中
+                .addLast(new IdleStateHandler(0, 30, 0))
 
 
                 //心跳解码
                 //心跳解码
                 //.addLast(new HeartbeatEncode())
                 //.addLast(new HeartbeatEncode())

@@ -28,6 +28,9 @@ public class AppConfiguration {
     @Value("${cim.clear.route.request.url}")
     @Value("${cim.clear.route.request.url}")
     private String clearRouteUrl ;
     private String clearRouteUrl ;
 
 
+    @Value("${cim.heartbeat.time}")
+    private long heartBeatTime ;
+
     public String getClearRouteUrl() {
     public String getClearRouteUrl() {
         return clearRouteUrl;
         return clearRouteUrl;
     }
     }
@@ -67,4 +70,12 @@ public class AppConfiguration {
     public void setCimServerPort(int cimServerPort) {
     public void setCimServerPort(int cimServerPort) {
         this.cimServerPort = cimServerPort;
         this.cimServerPort = cimServerPort;
     }
     }
+
+    public long getHeartBeatTime() {
+        return heartBeatTime;
+    }
+
+    public void setHeartBeatTime(long heartBeatTime) {
+        this.heartBeatTime = heartBeatTime;
+    }
 }
 }

@@ -6,6 +6,7 @@ import com.crossoverjie.cim.common.exception.CIMException;
 import com.crossoverjie.cim.common.pojo.CIMUserInfo;
 import com.crossoverjie.cim.common.pojo.CIMUserInfo;
 import com.crossoverjie.cim.common.protocol.CIMRequestProto;
 import com.crossoverjie.cim.common.protocol.CIMRequestProto;
 import com.crossoverjie.cim.server.config.AppConfiguration;
 import com.crossoverjie.cim.server.config.AppConfiguration;
+import com.crossoverjie.cim.server.util.NettyAttrUtil;
 import com.crossoverjie.cim.server.util.SessionSocketHolder;
 import com.crossoverjie.cim.server.util.SessionSocketHolder;
 import com.crossoverjie.cim.server.util.SpringBeanFactory;
 import com.crossoverjie.cim.server.util.SpringBeanFactory;
 import io.netty.channel.*;
 import io.netty.channel.*;
@@ -50,26 +51,23 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
         if (evt instanceof IdleStateEvent) {
         if (evt instanceof IdleStateEvent) {
             IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
             IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
             if (idleStateEvent.state() == IdleState.READER_IDLE) {
             if (idleStateEvent.state() == IdleState.READER_IDLE) {
+                AppConfiguration configuration = SpringBeanFactory.getBean(AppConfiguration.class);
+                long heartBeatTime = configuration.getHeartBeatTime() * 1000;
+
 
 
                 //向客户端发送消息
                 //向客户端发送消息
                 CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",
                 CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",
                         CIMRequestProto.CIMReqProtocol.class);
                         CIMRequestProto.CIMReqProtocol.class);
-                ctx.writeAndFlush(heartBeat).addListeners(new ChannelFutureListener() {
-                    @Override
-                    public void operationComplete(ChannelFuture future) throws Exception {
-                        //下线客户端
-                        CIMUserInfo userInfo = SessionSocketHolder.getUserId((NioSocketChannel) future.channel());
-                        if (!future.isSuccess()) {
+                ctx.writeAndFlush(heartBeat).addListeners(ChannelFutureListener.CLOSE_ON_FAILURE);
 
 
-                            LOGGER.info("向客户端[{}]下发心跳失败",userInfo.getUserName());
-
-                            userOffLine(userInfo, (NioSocketChannel) future.channel());
-                            future.channel().close();
-                        }else {
-                            LOGGER.info("向客户端[{}]下发心跳成功",userInfo.getUserName());
-                        }
+                Long lastReadTime = NettyAttrUtil.getReaderTime(ctx.channel());
+                long now = System.currentTimeMillis();
+                if (lastReadTime != null && now - lastReadTime > heartBeatTime){
+                    CIMUserInfo userInfo = SessionSocketHolder.getUserId((NioSocketChannel) ctx.channel());
+                    LOGGER.warn("客户端[{}]心跳超时，需要关闭连接",userInfo.getUserName());
+                    userOffLine(userInfo, (NioSocketChannel) ctx.channel());
+                    ctx.channel().close();
                 }
                 }
-                });
             }
             }
         }
         }
         super.userEventTriggered(ctx, evt);
         super.userEventTriggered(ctx, evt);
@@ -132,6 +130,11 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
             LOGGER.info("客户端[{}]上线成功", msg.getReqMsg());
             LOGGER.info("客户端[{}]上线成功", msg.getReqMsg());
         }
         }
 
 
+        //心跳更新时间
+        if (msg.getType() == Constants.CommandType.PING){
+            NettyAttrUtil.updateReaderTime(ctx.channel(),System.currentTimeMillis());
+        }
+
     }
     }
 
 
 
 

@@ -25,7 +25,7 @@ public class CIMServerInitializer extends ChannelInitializer<Channel> {
     protected void initChannel(Channel ch) throws Exception {
     protected void initChannel(Channel ch) throws Exception {
 
 
         ch.pipeline()
         ch.pipeline()
-                //45 秒没有向客户端发送消息就发生心跳
+                //30 秒没有向客户端发送消息就发生心跳
                 .addLast(new IdleStateHandler(30, 0, 0))
                 .addLast(new IdleStateHandler(30, 0, 0))
                 // google Protobuf 编解码
                 // google Protobuf 编解码
                 .addLast(new ProtobufVarint32FrameDecoder())
                 .addLast(new ProtobufVarint32FrameDecoder())

+package com.crossoverjie.cim.server.util;
+
+import io.netty.channel.Channel;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ *         Date: 2019/1/9 00:57
+ * @since JDK 1.8
+ */
+public class NettyAttrUtil {
+
+    private static final AttributeKey<String> ATTR_KEY_READER_TIME = AttributeKey.valueOf("readerTime");
+
+
+    public static void updateReaderTime(Channel channel,Long time){
+        channel.attr(ATTR_KEY_READER_TIME).set(time.toString());
+    }
+
+    public static Long getReaderTime(Channel channel){
+        String value = getAttribute(channel, ATTR_KEY_READER_TIME);
+        return Long.valueOf(value) ;
+    }
+
+
+    private static String getAttribute(Channel channel, AttributeKey<String> key) {
+        Attribute<String> attr = channel.attr(key);
+        return attr.get();
+    }
+}

@@ -32,3 +32,6 @@ app.zk.root=/route
 
 
 # 清除路由信息
 # 清除路由信息
 cim.clear.route.request.url=http://localhost:8083/offLine
 cim.clear.route.request.url=http://localhost:8083/offLine
+
+# 检测多少秒没有收到客户端心跳后服务端关闭连接
+cim.heartbeat.time = 40

+package com.crossoverjie.cim.server.util;
+
+
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+public class NettyAttrUtilTest {
+
+    @Test
+    public void test() throws InterruptedException {
+        long heartbeat = 2 * 1000 ;
+
+        long now = System.currentTimeMillis();
+        TimeUnit.SECONDS.sleep(1);
+
+        long end = System.currentTimeMillis();
+
+        if ((end - now) > heartbeat){
+            System.out.println("超时");
+        }else {
+            System.out.println("没有超时");
+        }
+    }
+
+}