1:新增心跳处理器

customer-service/src/main/java/com/ym/im/core/MessageDecoder.java

@@ -8,6 +8,7 @@ package com.ym.im.core;
  */
 
 import com.ym.im.entity.ChatRecord;
+import com.ym.im.entity.HeartBeat;
 import com.ym.im.entity.MsgBody;
 import com.ym.im.util.JsonUtils;
 import io.netty.channel.ChannelHandler;
@@ -30,10 +31,16 @@ import java.util.List;
 @ChannelHandler.Sharable
 public class MessageDecoder extends MessageToMessageDecoder<TextWebSocketFrame> {
 
+    private final static String PING = "ping";
+
     @Override
     protected void decode(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame, List out) throws Exception {
 
         final String jsonStr = textWebSocketFrame.text();
+        if (PING.equals(jsonStr)) {
+            out.add(new HeartBeat());
+            return;
+        }
         MsgBody<ChatRecord> msgBody = new MsgBody<ChatRecord>().setCode(MsgBody.ERROR);
         try {
             msgBody = JsonUtils.json2Obj(jsonStr, MsgBody.class, ChatRecord.class);

customer-service/src/main/java/com/ym/im/core/WebSocketChannelInitializer.java

@@ -2,6 +2,7 @@ package com.ym.im.core;
 
 
 import com.ym.im.entity.base.NettyConstant;
+import com.ym.im.handler.HeartBeatServerHandler;
 import com.ym.im.handler.SingleChatHandler;
 import com.ym.im.handler.WebSocketHandshakerHandler;
 import io.netty.channel.ChannelInitializer;
@@ -9,6 +10,7 @@ import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.EventExecutorGroup;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -35,6 +37,9 @@ public class WebSocketChannelInitializer extends ChannelInitializer<NioSocketCha
     private SingleChatHandler singleChatHandler;
 
     @Autowired
+    private HeartBeatServerHandler heartBeatServerHandler;
+
+    @Autowired
     private WebSocketHandshakerHandler webSocketHandshakerHandler;
 
 
@@ -45,8 +50,10 @@ public class WebSocketChannelInitializer extends ChannelInitializer<NioSocketCha
                         new HttpObjectAggregator(NettyConstant.MAX_FRAME_LENGTH),
                         webSocketHandshakerHandler,
                         new WebSocketServerProtocolHandler(NettyConstant.CS),
+                        new IdleStateHandler(60, 0, 0),
                         messageDecoder,
                         messageEncoder)
-                .addLast(businessGroup, singleChatHandler);//复杂业务绑定businessGroup
+                .addLast(businessGroup, singleChatHandler)//复杂业务绑定businessGroup
+                .addLast(heartBeatServerHandler);
     }
 }

customer-service/src/main/java/com/ym/im/entity/HeartBeat.java

+package com.ym.im.entity;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @author: JJww
+ * @Date:2020/11/6
+ */
+@Data
+public class HeartBeat implements Serializable {
+}

customer-service/src/main/java/com/ym/im/handler/HeartBeatServerHandler.java

+package com.ym.im.handler;
+
+import com.ym.im.entity.HeartBeat;
+import com.ym.im.entity.base.ChannelAttributeKey;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author: JJww
+ * @Date:2020/11/6
+ */
+@Slf4j
+@Component
+@ChannelHandler.Sharable
+public class HeartBeatServerHandler extends BaseHandler<HeartBeat> {
+
+    private final static String PONG = "pong";
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleStateEvent) {
+            IdleStateEvent event = (IdleStateEvent) evt;
+            if (event.state() == IdleState.READER_IDLE) {
+                ctx.channel().close();
+                final Long userId = ctx.channel().attr(ChannelAttributeKey.ROLE_ID).get();
+                log.info("用户: " + userId + "  无心跳 ");
+            }
+        } else {
+            super.userEventTriggered(ctx, evt);
+        }
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, HeartBeat heartBeat) throws Exception {
+        ctx.channel().writeAndFlush(new TextWebSocketFrame(PONG));
+    }
+}