客户端可在控制台发送消息

 package com.crossoverjie.netty.action.client;
 
+import com.crossoverjie.netty.action.client.scanner.Scan;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -14,6 +16,8 @@ public class HeartbeatClientApplication implements CommandLineRunner{
 
 	private final static Logger LOGGER = LoggerFactory.getLogger(HeartbeatClientApplication.class);
 
+	@Autowired
+	private HeartbeatClient heartbeatClient ;
 
 
 	public static void main(String[] args) {
@@ -23,5 +27,9 @@ public class HeartbeatClientApplication implements CommandLineRunner{
 
 	@Override
 	public void run(String... args) throws Exception {
+		Scan scan = new Scan(heartbeatClient) ;
+		Thread thread = new Thread(scan);
+		thread.setName("scan-thread");
+		thread.start();
 	}
 }
\ No newline at end of file

 package com.crossoverjie.netty.action.client.handle;
 
+import com.crossoverjie.netty.action.common.protocol.BaseRequestProto;
 import com.crossoverjie.netty.action.common.protocol.BaseResponseProto;
+import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -17,23 +21,26 @@ public class EchoClientHandle extends SimpleChannelInboundHandler<BaseResponsePr
 
     private final static Logger LOGGER = LoggerFactory.getLogger(EchoClientHandle.class);
 
+    private final BaseRequestProto.RequestProtocol heart = BaseRequestProto.RequestProtocol.newBuilder()
+            .setRequestId(99999999)
+            .setReqMsg("ping")
+            .build();
 
 
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
 
-        /*if (evt instanceof IdleStateEvent){
+        if (evt instanceof IdleStateEvent){
             IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;
 
             if (idleStateEvent.state() == IdleState.WRITER_IDLE){
                 LOGGER.info("已经 10 秒没有发送信息！");
                 //向服务端发送消息
-                CustomProtocol heartBeat = SpringBeanFactory.getBean("heartBeat", CustomProtocol.class);
-                ctx.writeAndFlush(heartBeat).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;
+                ctx.writeAndFlush(heart).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;
             }
 
 
-        }*/
+        }
 
         super.userEventTriggered(ctx, evt);
     }

+package com.crossoverjie.netty.action.client.scanner;
+
+import com.crossoverjie.netty.action.client.HeartbeatClient;
+import com.crossoverjie.netty.action.client.vo.req.GoogleProtocolVO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Scanner;
+
+/**
+ * Function:
+ *
+ * @author crossoverJie
+ *         Date: 2018/12/21 16:44
+ * @since JDK 1.8
+ */
+public class Scan implements Runnable{
+
+    private final static Logger LOGGER = LoggerFactory.getLogger(Scan.class);
+
+    private HeartbeatClient heartbeatClient ;
+
+    public Scan(HeartbeatClient heartbeatClient) {
+        this.heartbeatClient = heartbeatClient;
+    }
+
+    @Override
+    public void run() {
+        Scanner sc = new Scanner(System.in);
+        String[] totalMsg ;
+        GoogleProtocolVO vo ;
+        while (true){
+            String msg = sc.nextLine() ;
+
+            totalMsg = msg.split(" ");
+
+            vo = new GoogleProtocolVO() ;
+            vo.setRequestId(Integer.parseInt(totalMsg[0]));
+            vo.setMsg(totalMsg[1]);
+            heartbeatClient.sendGoogleProtocolMsg(vo) ;
+            LOGGER.info("scan =[{}]",msg);
+        }
+    }
+}

@@ -2,9 +2,7 @@ package com.crossoverjie.netty.action.controller;
 
 import com.crossoverjie.netty.action.common.constant.Constants;
 import com.crossoverjie.netty.action.common.enums.StatusEnum;
-import com.crossoverjie.netty.action.common.pojo.CustomProtocol;
 import com.crossoverjie.netty.action.common.res.BaseResponse;
-import com.crossoverjie.netty.action.common.util.RandomUtil;
 import com.crossoverjie.netty.action.server.HeartBeatServer;
 import com.crossoverjie.netty.action.vo.req.SendMsgReqVO;
 import com.crossoverjie.netty.action.vo.res.SendMsgResVO;
@@ -16,8 +14,6 @@ import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.ResponseBody;
 
-import java.util.UUID;
-
 /**
  * Function:
  *
@@ -49,7 +45,7 @@ public class IndexController {
     @ResponseBody
     public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){
         BaseResponse<SendMsgResVO> res = new BaseResponse();
-        heartbeatClient.sendMsg(new CustomProtocol(sendMsgReqVO.getId(),sendMsgReqVO.getMsg())) ;
+        heartbeatClient.sendGoogleProtoMsg(sendMsgReqVO) ;
 
         counterService.increment(Constants.COUNTER_SERVER_PUSH_COUNT);
 

@@ -69,6 +69,6 @@ public class HeartBeatSimpleHandle extends SimpleChannelInboundHandler<BaseReque
         }
 
         //保存客户端与 Channel 之间的关系
-        //NettySocketHolder.put(CustomProtocolProtocol.getId(),(NioSocketChannel)ctx.channel()) ;
+        NettySocketHolder.put((long) msg.getRequestId(),(NioSocketChannel)ctx.channel()) ;
     }
 }

 package com.crossoverjie.netty.action.server;
 
 import com.alibaba.fastjson.JSON;
-import com.crossoverjie.netty.action.init.HeartbeatInitializer;
 import com.crossoverjie.netty.action.common.pojo.CustomProtocol;
+import com.crossoverjie.netty.action.common.protocol.BaseRequestProto;
+import com.crossoverjie.netty.action.init.HeartbeatInitializer;
 import com.crossoverjie.netty.action.util.NettySocketHolder;
+import com.crossoverjie.netty.action.vo.req.SendMsgReqVO;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.buffer.Unpooled;
-import io.netty.channel.*;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
@@ -91,4 +96,24 @@ public class HeartBeatServer {
         future.addListener((ChannelFutureListener) channelFuture ->
                 LOGGER.info("服务端手动发消息成功={}", JSON.toJSONString(customProtocol)));
     }
+
+    /**
+     * 发送 Google Protocol 编码消息
+     * @param sendMsgReqVO 消息
+     */
+    public void sendGoogleProtoMsg(SendMsgReqVO sendMsgReqVO){
+        NioSocketChannel socketChannel = NettySocketHolder.get(sendMsgReqVO.getId());
+
+        if (null == socketChannel) {
+            throw new NullPointerException("没有[" + sendMsgReqVO.getId() + "]的socketChannel");
+        }
+        BaseRequestProto.RequestProtocol protocol = BaseRequestProto.RequestProtocol.newBuilder()
+                .setRequestId((int) sendMsgReqVO.getId())
+                .setReqMsg(sendMsgReqVO.getMsg())
+                .build();
+
+        ChannelFuture future = socketChannel.writeAndFlush(protocol);
+        future.addListener((ChannelFutureListener) channelFuture ->
+                LOGGER.info("服务端手动发送 Google Protocol 成功={}", sendMsgReqVO.toString()));
+    }
 }

@@ -14,7 +14,7 @@ logging.level.root=info
 app.zk.switch=true
 
 # zk 地址
-app.zk.addr=127.0.0.1:2181
+app.zk.addr=47.98.194.60:2181
 
 # zk 注册根节点
 app.zk.root=/route
\ No newline at end of file