Introducing new features.后端和客户端通信

@@ -9,15 +9,15 @@ package com.crossoverjie.netty.action.common.pojo;
  */
 public class CustomProtocol {
 
-    private long header ;
+    private long id ;
     private String content ;
 
-    public long getHeader() {
-        return header;
+    public long getId() {
+        return id;
     }
 
-    public void setHeader(long header) {
-        this.header = header;
+    public void setId(long id) {
+        this.id = id;
     }
 
     public String getContent() {
@@ -28,18 +28,18 @@ public class CustomProtocol {
         this.content = content;
     }
 
-    public CustomProtocol(long header, String content) {
-        this.header = header;
-        this.content = content;
+    public CustomProtocol() {
     }
 
-    public CustomProtocol() {
+    public CustomProtocol(long id, String content) {
+        this.id = id;
+        this.content = content;
     }
 
     @Override
     public String toString() {
         return "CustomProtocol{" +
-                "header=" + header +
+                "id=" + id +
                 ", content='" + content + '\'' +
                 '}';
     }

@@ -38,7 +38,7 @@ public class IndexController {
     @ResponseBody
     public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){
         BaseResponse<SendMsgResVO> res = new BaseResponse();
-        heartbeatClient.sendMsg(new CustomProtocol(RandomUtil.getRandom(),sendMsgReqVO.getMsg())) ;
+        heartbeatClient.sendMsg(new CustomProtocol(sendMsgReqVO.getId(),sendMsgReqVO.getMsg())) ;
 
         SendMsgResVO sendMsgResVO = new SendMsgResVO() ;
         sendMsgResVO.setMsg("OK") ;

@@ -16,7 +16,7 @@ public class HeartbeatEncode extends MessageToByteEncoder<CustomProtocol> {
     @Override
     protected void encode(ChannelHandlerContext ctx, CustomProtocol msg, ByteBuf out) throws Exception {
 
-        out.writeLong(msg.getHeader()) ;
+        out.writeLong(msg.getId()) ;
         out.writeBytes(msg.getContent().getBytes()) ;
 
     }

@@ -18,6 +18,10 @@ public class SendMsgReqVO extends BaseRequest {
     @ApiModelProperty(required = true, value = "msg", example = "hello")
     private String msg ;
 
+    @NotNull(message = "id 不能为空")
+    @ApiModelProperty(required = true, value = "id", example = "11")
+    private long id ;
+
     public String getMsg() {
         return msg;
     }
@@ -26,4 +30,11 @@ public class SendMsgReqVO extends BaseRequest {
         this.msg = msg;
     }
 
+    public long getId() {
+        return id;
+    }
+
+    public void setId(long id) {
+        this.id = id;
+    }
 }

@@ -40,7 +40,7 @@ public class IndexController {
     @ResponseBody
     public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){
         BaseResponse<SendMsgResVO> res = new BaseResponse();
-        heartbeatClient.sendMsg(new CustomProtocol(RandomUtil.getRandom(),sendMsgReqVO.getMsg())) ;
+        heartbeatClient.sendMsg(new CustomProtocol(sendMsgReqVO.getId(),sendMsgReqVO.getMsg())) ;
 
         SendMsgResVO sendMsgResVO = new SendMsgResVO() ;
         sendMsgResVO.setMsg("OK") ;

@@ -18,13 +18,13 @@ public class HeartbeatDecoder extends ByteToMessageDecoder {
     @Override
     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
 
-        long header = in.readLong() ;
+        long id = in.readLong() ;
         byte[] bytes = new byte[in.readableBytes()] ;
         in.readBytes(bytes) ;
         String content = new String(bytes) ;
 
         CustomProtocol customProtocol = new CustomProtocol() ;
-        customProtocol.setHeader(header);
+        customProtocol.setId(id);
         customProtocol.setContent(content) ;
         out.add(customProtocol) ;
 

 package com.crossoverjie.netty.action.handle;
 
 import com.crossoverjie.netty.action.common.pojo.CustomProtocol;
+import com.crossoverjie.netty.action.common.util.RandomUtil;
+import com.crossoverjie.netty.action.util.NettySocketHolder;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.timeout.IdleState;
 import io.netty.handler.timeout.IdleStateEvent;
 import io.netty.util.CharsetUtil;
@@ -32,7 +37,7 @@ public class HeartBeatSimpleHandle extends SimpleChannelInboundHandler<CustomPro
             if (idleStateEvent.state() == IdleState.READER_IDLE){
                 LOGGER.info("已经5秒没有收到信息！");
                 //向客户端发送消息
-                CustomProtocol customProtocol = new CustomProtocol(12345L,"pong") ;
+                CustomProtocol customProtocol = new CustomProtocol(RandomUtil.getRandom(),"pong") ;
                 ctx.writeAndFlush(Unpooled.copiedBuffer(customProtocol.toString(), CharsetUtil.UTF_8))
                         .addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;
             }
@@ -44,16 +49,9 @@ public class HeartBeatSimpleHandle extends SimpleChannelInboundHandler<CustomPro
     }
 
     @Override
-    public void channelActive(ChannelHandlerContext ctx) throws Exception {
-    }
-
-    @Override
     protected void channelRead0(ChannelHandlerContext ctx, CustomProtocol customProtocol) throws Exception {
         LOGGER.info("customProtocol={}", customProtocol);
 
-        //手动处理数据并返回
-        customProtocol.setHeader(customProtocol.getHeader() + 1000);
-        customProtocol.setContent(customProtocol.getContent() + "asdfg");
-        ctx.writeAndFlush(Unpooled.copiedBuffer(customProtocol.toString(), CharsetUtil.UTF_8));
+        NettySocketHolder.put(customProtocol.getId(),(NioSocketChannel)ctx.channel()) ;
     }
 }

@@ -3,11 +3,15 @@ package com.crossoverjie.netty.action.server;
 import com.alibaba.fastjson.JSON;
 import com.crossoverjie.netty.action.channel.init.HeartbeatInitializer;
 import com.crossoverjie.netty.action.common.pojo.CustomProtocol;
+import com.crossoverjie.netty.action.util.NettySocketHolder;
 import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.CharsetUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -78,8 +82,13 @@ public class HeartBeatServer {
      * @param customProtocol
      */
     public void sendMsg(CustomProtocol customProtocol){
-        ChannelFuture future = channel.writeAndFlush(customProtocol);
+        NioSocketChannel socketChannel = NettySocketHolder.get(customProtocol.getId());
 
+        if (null == socketChannel){
+            throw new NullPointerException("没有["+customProtocol.getId()+"]的socketChannel") ;
+        }
+
+        ChannelFuture future = socketChannel.writeAndFlush(Unpooled.copiedBuffer(customProtocol.toString(), CharsetUtil.UTF_8));
         future.addListener((ChannelFutureListener) channelFuture ->
                 LOGGER.info("服务端手动发消息成功={}", JSON.toJSONString(customProtocol)));
     }

+package com.crossoverjie.netty.action.util;
+
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ *         Date: 22/05/2018 18:33
+ * @since JDK 1.8
+ */
+public class NettySocketHolder {
+    private static final Map<Long,NioSocketChannel> MAP = new ConcurrentHashMap<>(16) ;
+
+    public static void put(Long id,NioSocketChannel socketChannel){
+        MAP.put(id,socketChannel) ;
+    }
+
+    public static NioSocketChannel get(Long id) {
+        return MAP.get(id);
+    }
+}

+package com.crossoverjie.netty.action.vo.req;
+
+import com.crossoverjie.netty.action.common.req.BaseRequest;
+import io.swagger.annotations.ApiModelProperty;
+
+import javax.validation.constraints.NotNull;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ *         Date: 2018/05/21 15:56
+ * @since JDK 1.8
+ */
+public class SendMsgReqVO extends BaseRequest {
+
+    @NotNull(message = "msg 不能为空")
+    @ApiModelProperty(required = true, value = "msg", example = "hello")
+    private String msg ;
+
+    @NotNull(message = "id 不能为空")
+    @ApiModelProperty(required = true, value = "id", example = "11")
+    private long id ;
+
+    public String getMsg() {
+        return msg;
+    }
+
+    public void setMsg(String msg) {
+        this.msg = msg;
+    }
+
+    public long getId() {
+        return id;
+    }
+
+    public void setId(long id) {
+        this.id = id;
+    }
+}

+package com.crossoverjie.netty.action.vo.res;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ *         Date: 2017/6/26 15:43
+ * @since JDK 1.8
+ */
+public class SendMsgResVO {
+    private String msg ;
+
+    public String getMsg() {
+        return msg;
+    }
+
+    public void setMsg(String msg) {
+        this.msg = msg;
+    }
+}