open-soure / cim · 提交

转到一个项目

作者 crossoverJie
提交者 GitHub
提交 5b698a8b3e32109ba90c8057b3877654be930200 2 个父辈 61b008a6 d30eeec0

Merge pull request #26 from crossoverJie/cim-1.0.2 

cim 1.0.2
内嵌 并排对比
正在显示 33 个修改的文件 包含 591 行增加 和 204 行删除
@@ -73,12 +73,6 @@ @@ -73,12 +73,6 @@
73 </dependency> 73 </dependency>
74 74
75 <dependency> 75 <dependency>
76 - <groupId>io.netty</groupId>  
77 - <artifactId>netty-all</artifactId>  
78 - <version>${netty.version}</version>  
79 - </dependency>  
80 -  
81 - <dependency>  
82 <groupId>junit</groupId> 76 <groupId>junit</groupId>
83 <artifactId>junit</artifactId> 77 <artifactId>junit</artifactId>
84 </dependency> 78 </dependency>
1 package com.crossoverjie.cim.client; 1 package com.crossoverjie.cim.client;
2 2
3 import com.crossoverjie.cim.client.scanner.Scan; 3 import com.crossoverjie.cim.client.scanner.Scan;
  4 +import com.crossoverjie.cim.client.service.impl.ClientInfo;
4 import org.slf4j.Logger; 5 import org.slf4j.Logger;
5 import org.slf4j.LoggerFactory; 6 import org.slf4j.LoggerFactory;
  7 +import org.springframework.beans.factory.annotation.Autowired;
6 import org.springframework.boot.CommandLineRunner; 8 import org.springframework.boot.CommandLineRunner;
7 import org.springframework.boot.SpringApplication; 9 import org.springframework.boot.SpringApplication;
8 import org.springframework.boot.autoconfigure.SpringBootApplication; 10 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -15,6 +17,8 @@ public class CIMClientApplication implements CommandLineRunner{ @@ -15,6 +17,8 @@ public class CIMClientApplication implements CommandLineRunner{
15 17
16 private final static Logger LOGGER = LoggerFactory.getLogger(CIMClientApplication.class); 18 private final static Logger LOGGER = LoggerFactory.getLogger(CIMClientApplication.class);
17 19
  20 + @Autowired
  21 + private ClientInfo clientInfo ;
18 public static void main(String[] args) { 22 public static void main(String[] args) {
19 SpringApplication.run(CIMClientApplication.class, args); 23 SpringApplication.run(CIMClientApplication.class, args);
20 LOGGER.info("启动 Client 服务成功"); 24 LOGGER.info("启动 Client 服务成功");
@@ -26,5 +30,6 @@ public class CIMClientApplication implements CommandLineRunner{ @@ -26,5 +30,6 @@ public class CIMClientApplication implements CommandLineRunner{
26 Thread thread = new Thread(scan); 30 Thread thread = new Thread(scan);
27 thread.setName("scan-thread"); 31 thread.setName("scan-thread");
28 thread.start(); 32 thread.start();
  33 + clientInfo.saveStartDate();
29 } 34 }
30 } 35 }
1 package com.crossoverjie.cim.client.client; 1 package com.crossoverjie.cim.client.client;
2 2
  3 +import com.crossoverjie.cim.client.config.AppConfiguration;
3 import com.crossoverjie.cim.client.init.CIMClientHandleInitializer; 4 import com.crossoverjie.cim.client.init.CIMClientHandleInitializer;
  5 +import com.crossoverjie.cim.client.service.MsgHandle;
4 import com.crossoverjie.cim.client.service.RouteRequest; 6 import com.crossoverjie.cim.client.service.RouteRequest;
  7 +import com.crossoverjie.cim.client.service.impl.ClientInfo;
5 import com.crossoverjie.cim.client.vo.req.GoogleProtocolVO; 8 import com.crossoverjie.cim.client.vo.req.GoogleProtocolVO;
6 import com.crossoverjie.cim.client.vo.req.LoginReqVO; 9 import com.crossoverjie.cim.client.vo.req.LoginReqVO;
7 import com.crossoverjie.cim.client.vo.res.CIMServerResVO; 10 import com.crossoverjie.cim.client.vo.res.CIMServerResVO;
@@ -16,6 +19,7 @@ import io.netty.channel.EventLoopGroup; @@ -16,6 +19,7 @@ import io.netty.channel.EventLoopGroup;
16 import io.netty.channel.nio.NioEventLoopGroup; 19 import io.netty.channel.nio.NioEventLoopGroup;
17 import io.netty.channel.socket.SocketChannel; 20 import io.netty.channel.socket.SocketChannel;
18 import io.netty.channel.socket.nio.NioSocketChannel; 21 import io.netty.channel.socket.nio.NioSocketChannel;
  22 +import io.netty.util.concurrent.DefaultThreadFactory;
19 import org.slf4j.Logger; 23 import org.slf4j.Logger;
20 import org.slf4j.LoggerFactory; 24 import org.slf4j.LoggerFactory;
21 import org.springframework.beans.factory.annotation.Autowired; 25 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,7 +40,7 @@ public class CIMClient { @@ -36,7 +40,7 @@ public class CIMClient {
36 40
37 private final static Logger LOGGER = LoggerFactory.getLogger(CIMClient.class); 41 private final static Logger LOGGER = LoggerFactory.getLogger(CIMClient.class);
38 42
39 - private EventLoopGroup group = new NioEventLoopGroup(); 43 + private EventLoopGroup group = new NioEventLoopGroup(0, new DefaultThreadFactory("cim-work"));
40 44
41 @Value("${cim.user.id}") 45 @Value("${cim.user.id}")
42 private long userId; 46 private long userId;
@@ -49,6 +53,20 @@ public class CIMClient { @@ -49,6 +53,20 @@ public class CIMClient {
49 @Autowired 53 @Autowired
50 private RouteRequest routeRequest; 54 private RouteRequest routeRequest;
51 55
  56 + @Autowired
  57 + private AppConfiguration configuration;
  58 +
  59 + @Autowired
  60 + private MsgHandle msgHandle;
  61 +
  62 + @Autowired
  63 + private ClientInfo clientInfo;
  64 +
  65 + /**
  66 + * 重试次数
  67 + */
  68 + private int errorCount;
  69 +
52 @PostConstruct 70 @PostConstruct
53 public void start() throws Exception { 71 public void start() throws Exception {
54 72
@@ -70,14 +88,25 @@ public class CIMClient { @@ -70,14 +88,25 @@ public class CIMClient {
70 * @param cimServer 88 * @param cimServer
71 * @throws InterruptedException 89 * @throws InterruptedException
72 */ 90 */
73 - private void startClient(CIMServerResVO.ServerInfo cimServer) throws InterruptedException { 91 + private void startClient(CIMServerResVO.ServerInfo cimServer) {
74 Bootstrap bootstrap = new Bootstrap(); 92 Bootstrap bootstrap = new Bootstrap();
75 bootstrap.group(group) 93 bootstrap.group(group)
76 .channel(NioSocketChannel.class) 94 .channel(NioSocketChannel.class)
77 .handler(new CIMClientHandleInitializer()) 95 .handler(new CIMClientHandleInitializer())
78 ; 96 ;
79 97
80 - ChannelFuture future = bootstrap.connect(cimServer.getIp(), cimServer.getCimServerPort()).sync(); 98 + ChannelFuture future = null;
  99 + try {
  100 + future = bootstrap.connect(cimServer.getIp(), cimServer.getCimServerPort()).sync();
  101 + } catch (InterruptedException e) {
  102 + errorCount++;
  103 +
  104 + if (errorCount >= configuration.getErrorCount()) {
  105 + LOGGER.error("链接失败次数达到上限[{}]次", errorCount);
  106 + msgHandle.shutdown();
  107 + }
  108 + LOGGER.error("连接失败", e);
  109 + }
81 if (future.isSuccess()) { 110 if (future.isSuccess()) {
82 LOGGER.info("启动 cim client 成功"); 111 LOGGER.info("启动 cim client 成功");
83 } 112 }
@@ -90,10 +119,26 @@ public class CIMClient { @@ -90,10 +119,26 @@ public class CIMClient {
90 * @return 路由服务器信息 119 * @return 路由服务器信息
91 * @throws Exception 120 * @throws Exception
92 */ 121 */
93 - private CIMServerResVO.ServerInfo userLogin() throws Exception { 122 + private CIMServerResVO.ServerInfo userLogin() {
94 LoginReqVO loginReqVO = new LoginReqVO(userId, userName); 123 LoginReqVO loginReqVO = new LoginReqVO(userId, userName);
95 - CIMServerResVO.ServerInfo cimServer = routeRequest.getCIMServer(loginReqVO); 124 + CIMServerResVO.ServerInfo cimServer = null;
  125 + try {
  126 + cimServer = routeRequest.getCIMServer(loginReqVO);
  127 +
  128 + //保存系统信息
  129 + clientInfo.saveServiceInfo(cimServer.getIp() + ":" + cimServer.getCimServerPort())
  130 + .saveUserInfo(userId, userName);
  131 +
96 LOGGER.info("cimServer=[{}]", cimServer.toString()); 132 LOGGER.info("cimServer=[{}]", cimServer.toString());
  133 + } catch (Exception e) {
  134 + errorCount++;
  135 +
  136 + if (errorCount >= configuration.getErrorCount()) {
  137 + LOGGER.error("重连次数达到上限[{}]次", errorCount);
  138 + msgHandle.shutdown();
  139 + }
  140 + LOGGER.error("登录失败", e);
  141 + }
97 return cimServer; 142 return cimServer;
98 } 143 }
99 144
@@ -145,11 +190,25 @@ public class CIMClient { @@ -145,11 +190,25 @@ public class CIMClient {
145 190
146 } 191 }
147 192
  193 +
  194 + public void reconnect() throws Exception {
  195 + if (channel != null && channel.isActive()) {
  196 + return;
  197 + }
  198 + //首先清除路由信息，下线
  199 + routeRequest.offLine();
  200 +
  201 + LOGGER.info("开始重连。。");
  202 + start();
  203 + LOGGER.info("重连成功！！");
  204 + }
  205 +
148 /** 206 /**
149 * 关闭 207 * 关闭
  208 + *
150 * @throws InterruptedException 209 * @throws InterruptedException
151 */ 210 */
152 public void close() throws InterruptedException { 211 public void close() throws InterruptedException {
153 - channel.close() ; 212 + channel.close();
154 } 213 }
155 } 214 }
@@ -22,6 +22,15 @@ public class AppConfiguration { @@ -22,6 +22,15 @@ public class AppConfiguration {
22 @Value("${cim.msg.logger.path}") 22 @Value("${cim.msg.logger.path}")
23 private String msgLoggerPath ; 23 private String msgLoggerPath ;
24 24
  25 + @Value("${cim.clear.route.request.url}")
  26 + private String clearRouteUrl ;
  27 +
  28 + @Value("${cim.heartbeat.time}")
  29 + private long heartBeatTime ;
  30 +
  31 + @Value("${cim.reconnect.count}")
  32 + private int errorCount ;
  33 +
25 public Long getUserId() { 34 public Long getUserId() {
26 return userId; 35 return userId;
27 } 36 }
@@ -45,4 +54,30 @@ public class AppConfiguration { @@ -45,4 +54,30 @@ public class AppConfiguration {
45 public void setMsgLoggerPath(String msgLoggerPath) { 54 public void setMsgLoggerPath(String msgLoggerPath) {
46 this.msgLoggerPath = msgLoggerPath; 55 this.msgLoggerPath = msgLoggerPath;
47 } 56 }
  57 +
  58 +
  59 + public long getHeartBeatTime() {
  60 + return heartBeatTime;
  61 + }
  62 +
  63 + public void setHeartBeatTime(long heartBeatTime) {
  64 + this.heartBeatTime = heartBeatTime;
  65 + }
  66 +
  67 +
  68 + public String getClearRouteUrl() {
  69 + return clearRouteUrl;
  70 + }
  71 +
  72 + public void setClearRouteUrl(String clearRouteUrl) {
  73 + this.clearRouteUrl = clearRouteUrl;
  74 + }
  75 +
  76 + public int getErrorCount() {
  77 + return errorCount;
  78 + }
  79 +
  80 + public void setErrorCount(int errorCount) {
  81 + this.errorCount = errorCount;
  82 + }
48 } 83 }
@@ -82,6 +82,17 @@ public class BeanConfig { @@ -82,6 +82,17 @@ public class BeanConfig {
82 return productExecutor ; 82 return productExecutor ;
83 } 83 }
84 84
  85 +
  86 + @Bean("scheduledTask")
  87 + public ScheduledExecutorService buildSchedule(){
  88 + ThreadFactory sche = new ThreadFactoryBuilder()
  89 + .setNameFormat("scheduled-%d")
  90 + .setDaemon(true)
  91 + .build();
  92 + ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1,sche) ;
  93 + return scheduledExecutorService ;
  94 + }
  95 +
85 /** 96 /**
86 * 回调 bean 97 * 回调 bean
87 * @return 98 * @return
1 package com.crossoverjie.cim.client.handle; 1 package com.crossoverjie.cim.client.handle;
2 2
  3 +import com.crossoverjie.cim.client.thread.ReConnectJob;
3 import com.crossoverjie.cim.client.util.SpringBeanFactory; 4 import com.crossoverjie.cim.client.util.SpringBeanFactory;
4 import com.crossoverjie.cim.common.constant.Constants; 5 import com.crossoverjie.cim.common.constant.Constants;
5 import com.crossoverjie.cim.common.protocol.CIMRequestProto; 6 import com.crossoverjie.cim.common.protocol.CIMRequestProto;
6 import com.crossoverjie.cim.common.protocol.CIMResponseProto; 7 import com.crossoverjie.cim.common.protocol.CIMResponseProto;
  8 +import com.crossoverjie.cim.common.util.NettyAttrUtil;
7 import io.netty.channel.ChannelFutureListener; 9 import io.netty.channel.ChannelFutureListener;
8 import io.netty.channel.ChannelHandler; 10 import io.netty.channel.ChannelHandler;
9 import io.netty.channel.ChannelHandlerContext; 11 import io.netty.channel.ChannelHandlerContext;
@@ -13,7 +15,9 @@ import io.netty.handler.timeout.IdleStateEvent; @@ -13,7 +15,9 @@ import io.netty.handler.timeout.IdleStateEvent;
13 import org.slf4j.Logger; 15 import org.slf4j.Logger;
14 import org.slf4j.LoggerFactory; 16 import org.slf4j.LoggerFactory;
15 17
  18 +import java.util.concurrent.ScheduledExecutorService;
16 import java.util.concurrent.ThreadPoolExecutor; 19 import java.util.concurrent.ThreadPoolExecutor;
  20 +import java.util.concurrent.TimeUnit;
17 21
18 /** 22 /**
19 * Function: 23 * Function:
@@ -31,6 +35,8 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt @@ -31,6 +35,8 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
31 35
32 private ThreadPoolExecutor threadPoolExecutor ; 36 private ThreadPoolExecutor threadPoolExecutor ;
33 37
  38 + private ScheduledExecutorService scheduledExecutorService ;
  39 +
34 40
35 @Override 41 @Override
36 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 42 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
@@ -38,12 +44,18 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt @@ -38,12 +44,18 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
38 if (evt instanceof IdleStateEvent){ 44 if (evt instanceof IdleStateEvent){
39 IdleStateEvent idleStateEvent = (IdleStateEvent) evt ; 45 IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;
40 46
  47 + //LOGGER.info("定时检测服务端是否存活");
  48 +
41 if (idleStateEvent.state() == IdleState.WRITER_IDLE){ 49 if (idleStateEvent.state() == IdleState.WRITER_IDLE){
42 CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat", 50 CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",
43 CIMRequestProto.CIMReqProtocol.class); 51 CIMRequestProto.CIMReqProtocol.class);
44 - ctx.writeAndFlush(heartBeat).addListeners(ChannelFutureListener.CLOSE_ON_FAILURE) ; 52 + ctx.writeAndFlush(heartBeat).addListeners((ChannelFutureListener) future -> {
  53 + if (!future.isSuccess()) {
  54 + LOGGER.error("IO error,close Channel");
  55 + future.channel().close();
  56 + }
  57 + }) ;
45 } 58 }
46 -  
47 59
48 } 60 }
49 61
@@ -58,10 +70,24 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt @@ -58,10 +70,24 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
58 } 70 }
59 71
60 @Override 72 @Override
61 - protected void channelRead0(ChannelHandlerContext channelHandlerContext, CIMResponseProto.CIMResProtocol msg) throws Exception { 73 + public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  74 + LOGGER.info("客户端断开了，重新连接！");
  75 +
  76 + if (scheduledExecutorService == null){
  77 + scheduledExecutorService = SpringBeanFactory.getBean("scheduledTask",ScheduledExecutorService.class) ;
  78 + }
  79 + // TODO: 2019-01-22 后期可以改为不用定时任务，连上后就关闭任务 节省性能。
  80 + scheduledExecutorService.scheduleAtFixedRate(new ReConnectJob(ctx),0,10, TimeUnit.SECONDS) ;
  81 + }
  82 +
  83 + @Override
  84 + protected void channelRead0(ChannelHandlerContext ctx, CIMResponseProto.CIMResProtocol msg) throws Exception {
62 85
63 - //从服务端收到消息时被调用  
64 - //LOGGER.info("客户端收到消息={}",in.toString(CharsetUtil.UTF_8)) ; 86 + //心跳更新时间
  87 + if (msg.getType() == Constants.CommandType.PING){
  88 + //LOGGER.info("收到服务端心跳！！！");
  89 + NettyAttrUtil.updateReaderTime(ctx.channel(),System.currentTimeMillis());
  90 + }
65 91
66 if (msg.getType() != Constants.CommandType.PING) { 92 if (msg.getType() != Constants.CommandType.PING) {
67 //回调消息 93 //回调消息
@@ -70,6 +96,10 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt @@ -70,6 +96,10 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
70 LOGGER.info(msg.getResMsg()); 96 LOGGER.info(msg.getResMsg());
71 } 97 }
72 98
  99 +
  100 +
  101 +
  102 +
73 } 103 }
74 104
75 /** 105 /**
@@ -24,8 +24,8 @@ public class CIMClientHandleInitializer extends ChannelInitializer<Channel> { @@ -24,8 +24,8 @@ public class CIMClientHandleInitializer extends ChannelInitializer<Channel> {
24 @Override 24 @Override
25 protected void initChannel(Channel ch) throws Exception { 25 protected void initChannel(Channel ch) throws Exception {
26 ch.pipeline() 26 ch.pipeline()
27 - //30 秒没发送消息 将IdleStateHandler 添加到 ChannelPipeline 中  
28 - .addLast(new IdleStateHandler(0, 30, 0)) 27 + //10 秒没发送消息 将IdleStateHandler 添加到 ChannelPipeline 中
  28 + .addLast(new IdleStateHandler(0, 10, 0))
29 29
30 //心跳解码 30 //心跳解码
31 //.addLast(new HeartbeatEncode()) 31 //.addLast(new HeartbeatEncode())
@@ -48,4 +48,10 @@ public interface MsgHandle { @@ -48,4 +48,10 @@ public interface MsgHandle {
48 * @return 是否应当跳过当前消息（包含了":" 就需要跳过） 48 * @return 是否应当跳过当前消息（包含了":" 就需要跳过）
49 */ 49 */
50 boolean innerCommand(String msg) ; 50 boolean innerCommand(String msg) ;
  51 +
  52 +
  53 + /**
  54 + * 关闭系统
  55 + */
  56 + void shutdown() ;
51 } 57 }
@@ -41,10 +41,13 @@ public interface RouteRequest { @@ -41,10 +41,13 @@ public interface RouteRequest {
41 CIMServerResVO.ServerInfo getCIMServer(LoginReqVO loginReqVO) throws Exception; 41 CIMServerResVO.ServerInfo getCIMServer(LoginReqVO loginReqVO) throws Exception;
42 42
43 /** 43 /**
44 - *  
45 - * @return 获取所有在线用户 44 + * 获取所有在线用户
  45 + * @return
  46 + * @throws Exception
46 */ 47 */
47 List<OnlineUsersResVO.DataBodyBean> onlineUsers()throws Exception ; 48 List<OnlineUsersResVO.DataBodyBean> onlineUsers()throws Exception ;
48 49
49 50
  51 + void offLine() ;
  52 +
50 } 53 }
  1 +package com.crossoverjie.cim.client.service.impl;
  2 +
  3 +import com.crossoverjie.cim.client.client.CIMClient;
  4 +import com.crossoverjie.cim.common.kit.HeartBeatHandler;
  5 +import io.netty.channel.ChannelHandlerContext;
  6 +import org.slf4j.Logger;
  7 +import org.slf4j.LoggerFactory;
  8 +import org.springframework.beans.factory.annotation.Autowired;
  9 +import org.springframework.stereotype.Service;
  10 +
  11 +/**
  12 + * Function:
  13 + *
  14 + * @author crossoverJie
  15 + * Date: 2019-01-20 17:16
  16 + * @since JDK 1.8
  17 + */
  18 +@Service
  19 +public class ClientHeartBeatHandlerImpl implements HeartBeatHandler {
  20 +
  21 + private final static Logger LOGGER = LoggerFactory.getLogger(ClientHeartBeatHandlerImpl.class);
  22 +
  23 + @Autowired
  24 + private CIMClient cimClient;
  25 +
  26 +
  27 + @Override
  28 + public void process(ChannelHandlerContext ctx) throws Exception {
  29 +
  30 + //重连
  31 + cimClient.reconnect();
  32 +
  33 + }
  34 +
  35 +
  36 +}
  1 +package com.crossoverjie.cim.client.service.impl;
  2 +
  3 +
  4 +import org.springframework.stereotype.Component;
  5 +
  6 +import java.util.Date;
  7 +
  8 +/**
  9 + * Function:
  10 + *
  11 + * @author crossoverJie
  12 + * Date: 2019-01-21 23:35
  13 + * @since JDK 1.8
  14 + */
  15 +@Component
  16 +public class ClientInfo {
  17 +
  18 + private Info info = new Info() ;
  19 +
  20 + public Info get(){
  21 + return info ;
  22 + }
  23 +
  24 + public ClientInfo saveUserInfo(long userId,String userName){
  25 + info.setUserId(userId);
  26 + info.setUserName(userName);
  27 + return this;
  28 + }
  29 +
  30 +
  31 + public ClientInfo saveServiceInfo(String serviceInfo){
  32 + info.setServiceInfo(serviceInfo);
  33 + return this;
  34 + }
  35 +
  36 + public ClientInfo saveStartDate(){
  37 + info.setStartDate(new Date());
  38 + return this;
  39 + }
  40 +
  41 + private class Info{
  42 + private String userName;
  43 + private long userId ;
  44 + private String serviceInfo ;
  45 + private Date startDate ;
  46 +
  47 + public Info() {
  48 + }
  49 +
  50 + public String getUserName() {
  51 + return userName;
  52 + }
  53 +
  54 + public void setUserName(String userName) {
  55 + this.userName = userName;
  56 + }
  57 +
  58 + public long getUserId() {
  59 + return userId;
  60 + }
  61 +
  62 + public void setUserId(long userId) {
  63 + this.userId = userId;
  64 + }
  65 +
  66 + public String getServiceInfo() {
  67 + return serviceInfo;
  68 + }
  69 +
  70 + public void setServiceInfo(String serviceInfo) {
  71 + this.serviceInfo = serviceInfo;
  72 + }
  73 +
  74 + public Date getStartDate() {
  75 + return startDate;
  76 + }
  77 +
  78 + public void setStartDate(Date startDate) {
  79 + this.startDate = startDate;
  80 + }
  81 + }
  82 +}
1 package com.crossoverjie.cim.client.service.impl; 1 package com.crossoverjie.cim.client.service.impl;
2 2
  3 +import com.alibaba.fastjson.JSON;
3 import com.crossoverjie.cim.client.client.CIMClient; 4 import com.crossoverjie.cim.client.client.CIMClient;
4 import com.crossoverjie.cim.client.config.AppConfiguration; 5 import com.crossoverjie.cim.client.config.AppConfiguration;
5 import com.crossoverjie.cim.client.service.MsgHandle; 6 import com.crossoverjie.cim.client.service.MsgHandle;
@@ -16,6 +17,7 @@ import org.slf4j.LoggerFactory; @@ -16,6 +17,7 @@ import org.slf4j.LoggerFactory;
16 import org.springframework.beans.factory.annotation.Autowired; 17 import org.springframework.beans.factory.annotation.Autowired;
17 import org.springframework.stereotype.Service; 18 import org.springframework.stereotype.Service;
18 19
  20 +import javax.annotation.Resource;
19 import java.util.List; 21 import java.util.List;
20 import java.util.Map; 22 import java.util.Map;
21 import java.util.concurrent.ThreadPoolExecutor; 23 import java.util.concurrent.ThreadPoolExecutor;
@@ -32,33 +34,37 @@ import java.util.concurrent.TimeUnit; @@ -32,33 +34,37 @@ import java.util.concurrent.TimeUnit;
32 public class MsgHandler implements MsgHandle { 34 public class MsgHandler implements MsgHandle {
33 private final static Logger LOGGER = LoggerFactory.getLogger(MsgHandler.class); 35 private final static Logger LOGGER = LoggerFactory.getLogger(MsgHandler.class);
34 @Autowired 36 @Autowired
35 - private RouteRequest routeRequest ; 37 + private RouteRequest routeRequest;
36 38
37 @Autowired 39 @Autowired
38 private AppConfiguration configuration; 40 private AppConfiguration configuration;
39 41
  42 + @Resource(name = "callBackThreadPool")
  43 + private ThreadPoolExecutor executor;
  44 +
40 @Autowired 45 @Autowired
41 - private ThreadPoolExecutor executor ; 46 + private CIMClient cimClient;
42 47
43 @Autowired 48 @Autowired
44 - private CIMClient cimClient ; 49 + private MsgLogger msgLogger;
45 50
46 @Autowired 51 @Autowired
47 - private MsgLogger msgLogger ; 52 + private ClientInfo clientInfo ;
48 53
49 - private boolean aiModel = false ; 54 + private boolean aiModel = false;
50 55
51 @Override 56 @Override
52 public void sendMsg(String msg) { 57 public void sendMsg(String msg) {
53 - if (aiModel){ 58 + if (aiModel) {
54 aiChat(msg); 59 aiChat(msg);
55 - }else { 60 + } else {
56 normalChat(msg); 61 normalChat(msg);
57 } 62 }
58 } 63 }
59 64
60 /** 65 /**
61 * 正常聊天 66 * 正常聊天
  67 + *
62 * @param msg 68 * @param msg
63 */ 69 */
64 private void normalChat(String msg) { 70 private void normalChat(String msg) {
@@ -72,7 +78,7 @@ public class MsgHandler implements MsgHandle { @@ -72,7 +78,7 @@ public class MsgHandler implements MsgHandle {
72 try { 78 try {
73 p2pChat(p2PReqVO); 79 p2pChat(p2PReqVO);
74 } catch (Exception e) { 80 } catch (Exception e) {
75 - LOGGER.error("Exception",e); 81 + LOGGER.error("Exception", e);
76 } 82 }
77 83
78 } else { 84 } else {
@@ -81,21 +87,22 @@ public class MsgHandler implements MsgHandle { @@ -81,21 +87,22 @@ public class MsgHandler implements MsgHandle {
81 try { 87 try {
82 groupChat(groupReqVO); 88 groupChat(groupReqVO);
83 } catch (Exception e) { 89 } catch (Exception e) {
84 - LOGGER.error("Exception",e); 90 + LOGGER.error("Exception", e);
85 } 91 }
86 } 92 }
87 } 93 }
88 94
89 /** 95 /**
90 * AI model 96 * AI model
  97 + *
91 * @param msg 98 * @param msg
92 */ 99 */
93 private void aiChat(String msg) { 100 private void aiChat(String msg) {
94 - msg = msg.replace("吗","") ;  
95 - msg = msg.replace("嘛","") ;  
96 - msg = msg.replace("?","!");  
97 - msg = msg.replace("？","!");  
98 - msg = msg.replace("你","我"); 101 + msg = msg.replace("吗", "");
  102 + msg = msg.replace("嘛", "");
  103 + msg = msg.replace("?", "!");
  104 + msg = msg.replace("？", "!");
  105 + msg = msg.replace("你", "我");
99 System.out.println("AI:\033[31;4m" + msg + "\033[0m"); 106 System.out.println("AI:\033[31;4m" + msg + "\033[0m");
100 } 107 }
101 108
@@ -113,7 +120,7 @@ public class MsgHandler implements MsgHandle { @@ -113,7 +120,7 @@ public class MsgHandler implements MsgHandle {
113 120
114 @Override 121 @Override
115 public boolean checkMsg(String msg) { 122 public boolean checkMsg(String msg) {
116 - if (StringUtil.isEmpty(msg)){ 123 + if (StringUtil.isEmpty(msg)) {
117 LOGGER.warn("不能发送空消息！"); 124 LOGGER.warn("不能发送空消息！");
118 return true; 125 return true;
119 } 126 }
@@ -123,41 +130,47 @@ public class MsgHandler implements MsgHandle { @@ -123,41 +130,47 @@ public class MsgHandler implements MsgHandle {
123 @Override 130 @Override
124 public boolean innerCommand(String msg) { 131 public boolean innerCommand(String msg) {
125 132
126 - if (msg.startsWith(":")){ 133 + // TODO: 2019-01-22 判断逻辑过多，需要重构。
  134 + if (msg.startsWith(":")) {
127 Map<String, String> allStatusCode = SystemCommandEnumType.getAllStatusCode(); 135 Map<String, String> allStatusCode = SystemCommandEnumType.getAllStatusCode();
128 136
129 - if (SystemCommandEnumType.QUIT.getCommandType().trim().equals(msg)){ 137 + if (SystemCommandEnumType.QUIT.getCommandType().trim().equals(msg)) {
130 //关闭系统 138 //关闭系统
131 shutdown(); 139 shutdown();
132 - } else if (SystemCommandEnumType.ALL.getCommandType().trim().equals(msg)){ 140 + } else if (SystemCommandEnumType.ALL.getCommandType().trim().equals(msg)) {
133 printAllCommand(allStatusCode); 141 printAllCommand(allStatusCode);
134 142
135 - } else if (SystemCommandEnumType.ONLINE_USER.getCommandType().toLowerCase().trim().equals(msg.toLowerCase())){ 143 + } else if (SystemCommandEnumType.ONLINE_USER.getCommandType().toLowerCase().trim().equals(msg.toLowerCase())) {
136 //打印在线用户 144 //打印在线用户
137 printOnlineUsers(); 145 printOnlineUsers();
138 146
139 - } else if (msg.startsWith(SystemCommandEnumType.QUERY.getCommandType().trim() + " ")){ 147 + } else if (msg.startsWith(SystemCommandEnumType.QUERY.getCommandType().trim() + " ")) {
140 //查询聊天记录 148 //查询聊天记录
141 queryChatHistory(msg); 149 queryChatHistory(msg);
142 - }else if (SystemCommandEnumType.AI.getCommandType().trim().equals(msg.toLowerCase())){ 150 + } else if (SystemCommandEnumType.AI.getCommandType().trim().equals(msg.toLowerCase())) {
143 //开启 AI 模式 151 //开启 AI 模式
144 - aiModel = true ; 152 + aiModel = true;
145 System.out.println("\033[31;4m" + "Hello,我是估值两亿的 AI 机器人！" + "\033[0m"); 153 System.out.println("\033[31;4m" + "Hello,我是估值两亿的 AI 机器人！" + "\033[0m");
146 - }else if (SystemCommandEnumType.QAI.getCommandType().trim().equals(msg.toLowerCase())){ 154 + } else if (SystemCommandEnumType.QAI.getCommandType().trim().equals(msg.toLowerCase())) {
147 //关闭 AI 模式 155 //关闭 AI 模式
148 - aiModel = false ; 156 + aiModel = false;
149 System.out.println("\033[31;4m" + "｡ﾟ(ﾟ´ω`ﾟ)ﾟ｡ AI 下线了！" + "\033[0m"); 157 System.out.println("\033[31;4m" + "｡ﾟ(ﾟ´ω`ﾟ)ﾟ｡ AI 下线了！" + "\033[0m");
150 - }else if (msg.startsWith(SystemCommandEnumType.PREFIX.getCommandType().trim() + " ")){ 158 + } else if (msg.startsWith(SystemCommandEnumType.PREFIX.getCommandType().trim() + " ")) {
151 //模糊匹配 159 //模糊匹配
152 prefixSearch(msg); 160 prefixSearch(msg);
153 - }else { 161 + } else if (SystemCommandEnumType.INFO.getCommandType().trim().equals(msg.toLowerCase())) {
  162 + LOGGER.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
  163 + LOGGER.info("client info=[{}]", JSON.toJSONString(clientInfo.get()));
  164 + LOGGER.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
  165 +
  166 + } else {
154 printAllCommand(allStatusCode); 167 printAllCommand(allStatusCode);
155 } 168 }
156 169
157 - return true ; 170 + return true;
158 171
159 - }else {  
160 - return false ; 172 + } else {
  173 + return false;
161 } 174 }
162 175
163 176
@@ -166,12 +179,13 @@ public class MsgHandler implements MsgHandle { @@ -166,12 +179,13 @@ public class MsgHandler implements MsgHandle {
166 179
167 /** 180 /**
168 * 模糊匹配 181 * 模糊匹配
  182 + *
169 * @param msg 183 * @param msg
170 */ 184 */
171 private void prefixSearch(String msg) { 185 private void prefixSearch(String msg) {
172 try { 186 try {
173 List<OnlineUsersResVO.DataBodyBean> onlineUsers = routeRequest.onlineUsers(); 187 List<OnlineUsersResVO.DataBodyBean> onlineUsers = routeRequest.onlineUsers();
174 - TrieTree trieTree = new TrieTree() ; 188 + TrieTree trieTree = new TrieTree();
175 for (OnlineUsersResVO.DataBodyBean onlineUser : onlineUsers) { 189 for (OnlineUsersResVO.DataBodyBean onlineUser : onlineUsers) {
176 trieTree.insert(onlineUser.getUserName()); 190 trieTree.insert(onlineUser.getUserName());
177 } 191 }
@@ -186,16 +200,17 @@ public class MsgHandler implements MsgHandle { @@ -186,16 +200,17 @@ public class MsgHandler implements MsgHandle {
186 } 200 }
187 201
188 } catch (Exception e) { 202 } catch (Exception e) {
189 - LOGGER.error("Exception" ,e); 203 + LOGGER.error("Exception", e);
190 } 204 }
191 } 205 }
192 206
193 /** 207 /**
194 * 查询聊天记录 208 * 查询聊天记录
  209 + *
195 * @param msg 210 * @param msg
196 */ 211 */
197 private void queryChatHistory(String msg) { 212 private void queryChatHistory(String msg) {
198 - String[] split = msg.split(" ") ; 213 + String[] split = msg.split(" ");
199 String res = msgLogger.query(split[1]); 214 String res = msgLogger.query(split[1]);
200 System.out.println(res); 215 System.out.println(res);
201 } 216 }
@@ -209,20 +224,22 @@ public class MsgHandler implements MsgHandle { @@ -209,20 +224,22 @@ public class MsgHandler implements MsgHandle {
209 224
210 LOGGER.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"); 225 LOGGER.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
211 for (OnlineUsersResVO.DataBodyBean onlineUser : onlineUsers) { 226 for (OnlineUsersResVO.DataBodyBean onlineUser : onlineUsers) {
212 - LOGGER.info("userId={}=====userName={}",onlineUser.getUserId(),onlineUser.getUserName()); 227 + LOGGER.info("userId={}=====userName={}", onlineUser.getUserId(), onlineUser.getUserName());
213 } 228 }
214 LOGGER.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"); 229 LOGGER.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
215 230
216 } catch (Exception e) { 231 } catch (Exception e) {
217 - LOGGER.error("Exception" ,e); 232 + LOGGER.error("Exception", e);
218 } 233 }
219 } 234 }
220 235
221 /** 236 /**
222 * 关闭系统 237 * 关闭系统
223 */ 238 */
224 - private void shutdown() { 239 + @Override
  240 + public void shutdown() {
225 LOGGER.info("系统关闭中。。。。"); 241 LOGGER.info("系统关闭中。。。。");
  242 + routeRequest.offLine();
226 msgLogger.stop(); 243 msgLogger.stop();
227 executor.shutdown(); 244 executor.shutdown();
228 try { 245 try {
@@ -231,7 +248,7 @@ public class MsgHandler implements MsgHandle { @@ -231,7 +248,7 @@ public class MsgHandler implements MsgHandle {
231 } 248 }
232 cimClient.close(); 249 cimClient.close();
233 } catch (InterruptedException e) { 250 } catch (InterruptedException e) {
234 - LOGGER.error("InterruptedException",e); 251 + LOGGER.error("InterruptedException", e);
235 } 252 }
236 System.exit(0); 253 System.exit(0);
237 } 254 }
@@ -45,7 +45,7 @@ public class RouteRequestImpl implements RouteRequest { @@ -45,7 +45,7 @@ public class RouteRequestImpl implements RouteRequest {
45 private String p2pRouteRequestUrl; 45 private String p2pRouteRequestUrl;
46 46
47 @Value("${cim.server.route.request.url}") 47 @Value("${cim.server.route.request.url}")
48 - private String serverRouteRequestUrl; 48 + private String serverRouteLoginUrl;
49 49
50 @Value("${cim.server.online.user.url}") 50 @Value("${cim.server.online.user.url}")
51 private String onlineUserUrl; 51 private String onlineUserUrl;
@@ -120,7 +120,7 @@ public class RouteRequestImpl implements RouteRequest { @@ -120,7 +120,7 @@ public class RouteRequestImpl implements RouteRequest {
120 RequestBody requestBody = RequestBody.create(mediaType,jsonObject.toString()); 120 RequestBody requestBody = RequestBody.create(mediaType,jsonObject.toString());
121 121
122 Request request = new Request.Builder() 122 Request request = new Request.Builder()
123 - .url(serverRouteRequestUrl) 123 + .url(serverRouteLoginUrl)
124 .post(requestBody) 124 .post(requestBody)
125 .build(); 125 .build();
126 126
@@ -178,4 +178,26 @@ public class RouteRequestImpl implements RouteRequest { @@ -178,4 +178,26 @@ public class RouteRequestImpl implements RouteRequest {
178 178
179 return onlineUsersResVO.getDataBody(); 179 return onlineUsersResVO.getDataBody();
180 } 180 }
  181 +
  182 + @Override
  183 + public void offLine() {
  184 + JSONObject jsonObject = new JSONObject();
  185 + jsonObject.put("userId", appConfiguration.getUserId());
  186 + jsonObject.put("msg", "offLine");
  187 + RequestBody requestBody = RequestBody.create(mediaType, jsonObject.toString());
  188 +
  189 + Request request = new Request.Builder()
  190 + .url(appConfiguration.getClearRouteUrl())
  191 + .post(requestBody)
  192 + .build();
  193 +
  194 + Response response = null;
  195 + try {
  196 + response = okHttpClient.newCall(request).execute();
  197 + } catch (IOException e) {
  198 + LOGGER.error("exception",e);
  199 + } finally {
  200 + response.body().close();
  201 + }
  202 + }
181 } 203 }
  1 +package com.crossoverjie.cim.client.thread;
  2 +
  3 +import com.crossoverjie.cim.client.service.impl.ClientHeartBeatHandlerImpl;
  4 +import com.crossoverjie.cim.client.util.SpringBeanFactory;
  5 +import com.crossoverjie.cim.common.kit.HeartBeatHandler;
  6 +import io.netty.channel.ChannelHandlerContext;
  7 +import org.slf4j.Logger;
  8 +import org.slf4j.LoggerFactory;
  9 +
  10 +/**
  11 + * Function:
  12 + *
  13 + * @author crossoverJie
  14 + * Date: 2019-01-20 21:35
  15 + * @since JDK 1.8
  16 + */
  17 +public class ReConnectJob implements Runnable {
  18 +
  19 + private final static Logger LOGGER = LoggerFactory.getLogger(ReConnectJob.class);
  20 +
  21 + private ChannelHandlerContext context ;
  22 +
  23 + private HeartBeatHandler heartBeatHandler ;
  24 +
  25 + public ReConnectJob(ChannelHandlerContext context) {
  26 + this.context = context;
  27 + this.heartBeatHandler = SpringBeanFactory.getBean(ClientHeartBeatHandlerImpl.class) ;
  28 + }
  29 +
  30 + @Override
  31 + public void run() {
  32 + try {
  33 + heartBeatHandler.process(context);
  34 + } catch (Exception e) {
  35 + LOGGER.error("Exception",e);
  36 + }
  37 + }
  38 +}
@@ -25,6 +25,8 @@ cim.server.route.request.url=http://45.78.28.220:8083/login @@ -25,6 +25,8 @@ cim.server.route.request.url=http://45.78.28.220:8083/login
25 # 在线用户 25 # 在线用户
26 cim.server.online.user.url=http://45.78.28.220:8083/onlineUser 26 cim.server.online.user.url=http://45.78.28.220:8083/onlineUser
27 27
  28 +# 清除路由信息
  29 +cim.clear.route.request.url=http://45.78.28.220:8083/offLine
28 30
29 ###=======本地模拟======### 31 ###=======本地模拟======###
30 ## 群发消息 32 ## 群发消息
@@ -39,6 +41,9 @@ cim.server.online.user.url=http://45.78.28.220:8083/onlineUser @@ -39,6 +41,9 @@ cim.server.online.user.url=http://45.78.28.220:8083/onlineUser
39 ## 在线用户 41 ## 在线用户
40 #cim.server.online.user.url=http://localhost:8083/onlineUser 42 #cim.server.online.user.url=http://localhost:8083/onlineUser
41 43
  44 +# 清除路由信息
  45 +#cim.clear.route.request.url=http://localhost:8083/offLine
  46 +
42 # 客户端唯一ID 47 # 客户端唯一ID
43 cim.user.id=1545574841528 48 cim.user.id=1545574841528
44 cim.user.userName=zhangsan 49 cim.user.userName=zhangsan
@@ -52,3 +57,9 @@ cim.callback.thread.pool.size = 2 @@ -52,3 +57,9 @@ cim.callback.thread.pool.size = 2
52 management.security.enabled=false 57 management.security.enabled=false
53 # SpringAdmin 地址 58 # SpringAdmin 地址
54 spring.boot.admin.url=http://127.0.0.1:8888 59 spring.boot.admin.url=http://127.0.0.1:8888
  60 +
  61 +# 检测多少秒没有收到服务端端心跳后重新登录获取连接
  62 +cim.heartbeat.time = 60
  63 +
  64 +# 客户端连接失败重连次数
  65 +cim.reconnect.count =3
@@ -38,5 +38,11 @@ @@ -38,5 +38,11 @@
38 <groupId>com.github.sgroschupf</groupId> 38 <groupId>com.github.sgroschupf</groupId>
39 <artifactId>zkclient</artifactId> 39 <artifactId>zkclient</artifactId>
40 </dependency> 40 </dependency>
  41 +
  42 + <dependency>
  43 + <groupId>io.netty</groupId>
  44 + <artifactId>netty-all</artifactId>
  45 + <version>${netty.version}</version>
  46 + </dependency>
41 </dependencies> 47 </dependencies>
42 </project> 48 </project>
@@ -20,7 +20,8 @@ public enum SystemCommandEnumType { @@ -20,7 +20,8 @@ public enum SystemCommandEnumType {
20 QUERY(":q ","【:q 关键字】查询聊天记录"), 20 QUERY(":q ","【:q 关键字】查询聊天记录"),
21 AI(":ai ","开启 AI 模式"), 21 AI(":ai ","开启 AI 模式"),
22 QAI(":qai ","关闭 AI 模式"), 22 QAI(":qai ","关闭 AI 模式"),
23 - PREFIX(":pu ","模糊匹配用户") 23 + PREFIX(":pu ","模糊匹配用户"),
  24 + INFO(":info ","获取客户端信息")
24 25
25 ; 26 ;
26 27
  1 +package com.crossoverjie.cim.common.kit;
  2 +
  3 +import io.netty.channel.ChannelHandlerContext;
  4 +
  5 +/**
  6 + * Function:
  7 + *
  8 + * @author crossoverJie
  9 + * Date: 2019-01-20 17:15
  10 + * @since JDK 1.8
  11 + */
  12 +public interface HeartBeatHandler {
  13 +
  14 + /**
  15 + * 处理心跳
  16 + * @param ctx
  17 + * @throws Exception
  18 + */
  19 + void process(ChannelHandlerContext ctx) throws Exception ;
  20 +}
1 -package com.crossoverjie.cim.server.util; 1 +package com.crossoverjie.cim.common.util;
2 2
3 import io.netty.channel.Channel; 3 import io.netty.channel.Channel;
4 import io.netty.util.Attribute; 4 import io.netty.util.Attribute;
@@ -174,6 +174,9 @@ public class AccountServiceRedisImpl implements AccountService { @@ -174,6 +174,9 @@ public class AccountServiceRedisImpl implements AccountService {
174 174
175 @Override 175 @Override
176 public void offLine(Long userId) throws Exception { 176 public void offLine(Long userId) throws Exception {
  177 +
  178 + // TODO: 2019-01-21 改为一个原子命令，以防数据一致性
  179 +
177 //删除路由 180 //删除路由
178 redisTemplate.delete(ROUTE_PREFIX + userId) ; 181 redisTemplate.delete(ROUTE_PREFIX + userId) ;
179 182
@@ -64,13 +64,6 @@ @@ -64,13 +64,6 @@
64 <artifactId>spring-boot-starter-actuator</artifactId> 64 <artifactId>spring-boot-starter-actuator</artifactId>
65 </dependency> 65 </dependency>
66 66
67 -  
68 - <dependency>  
69 - <groupId>io.netty</groupId>  
70 - <artifactId>netty-all</artifactId>  
71 - <version>${netty.version}</version>  
72 - </dependency>  
73 -  
74 <dependency> 67 <dependency>
75 <groupId>junit</groupId> 68 <groupId>junit</groupId>
76 <artifactId>junit</artifactId> 69 <artifactId>junit</artifactId>
@@ -51,7 +51,7 @@ public class BeanConfig { @@ -51,7 +51,7 @@ public class BeanConfig {
51 public CIMRequestProto.CIMReqProtocol heartBeat() { 51 public CIMRequestProto.CIMReqProtocol heartBeat() {
52 CIMRequestProto.CIMReqProtocol heart = CIMRequestProto.CIMReqProtocol.newBuilder() 52 CIMRequestProto.CIMReqProtocol heart = CIMRequestProto.CIMReqProtocol.newBuilder()
53 .setRequestId(0L) 53 .setRequestId(0L)
54 - .setReqMsg("ping") 54 + .setReqMsg("pong")
55 .setType(Constants.CommandType.PING) 55 .setType(Constants.CommandType.PING)
56 .build(); 56 .build();
57 return heart; 57 return heart;
@@ -3,13 +3,18 @@ package com.crossoverjie.cim.server.handle; @@ -3,13 +3,18 @@ package com.crossoverjie.cim.server.handle;
3 import com.alibaba.fastjson.JSONObject; 3 import com.alibaba.fastjson.JSONObject;
4 import com.crossoverjie.cim.common.constant.Constants; 4 import com.crossoverjie.cim.common.constant.Constants;
5 import com.crossoverjie.cim.common.exception.CIMException; 5 import com.crossoverjie.cim.common.exception.CIMException;
  6 +import com.crossoverjie.cim.common.kit.HeartBeatHandler;
6 import com.crossoverjie.cim.common.pojo.CIMUserInfo; 7 import com.crossoverjie.cim.common.pojo.CIMUserInfo;
7 import com.crossoverjie.cim.common.protocol.CIMRequestProto; 8 import com.crossoverjie.cim.common.protocol.CIMRequestProto;
  9 +import com.crossoverjie.cim.common.util.NettyAttrUtil;
8 import com.crossoverjie.cim.server.config.AppConfiguration; 10 import com.crossoverjie.cim.server.config.AppConfiguration;
9 -import com.crossoverjie.cim.server.util.NettyAttrUtil; 11 +import com.crossoverjie.cim.server.kit.ServerHeartBeatHandlerImpl;
10 import com.crossoverjie.cim.server.util.SessionSocketHolder; 12 import com.crossoverjie.cim.server.util.SessionSocketHolder;
11 import com.crossoverjie.cim.server.util.SpringBeanFactory; 13 import com.crossoverjie.cim.server.util.SpringBeanFactory;
12 -import io.netty.channel.*; 14 +import io.netty.channel.ChannelFutureListener;
  15 +import io.netty.channel.ChannelHandler;
  16 +import io.netty.channel.ChannelHandlerContext;
  17 +import io.netty.channel.SimpleChannelInboundHandler;
13 import io.netty.channel.socket.nio.NioSocketChannel; 18 import io.netty.channel.socket.nio.NioSocketChannel;
14 import io.netty.handler.timeout.IdleState; 19 import io.netty.handler.timeout.IdleState;
15 import io.netty.handler.timeout.IdleStateEvent; 20 import io.netty.handler.timeout.IdleStateEvent;
@@ -55,23 +60,11 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto @@ -55,23 +60,11 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
55 if (evt instanceof IdleStateEvent) { 60 if (evt instanceof IdleStateEvent) {
56 IdleStateEvent idleStateEvent = (IdleStateEvent) evt; 61 IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
57 if (idleStateEvent.state() == IdleState.READER_IDLE) { 62 if (idleStateEvent.state() == IdleState.READER_IDLE) {
58 - AppConfiguration configuration = SpringBeanFactory.getBean(AppConfiguration.class);  
59 - long heartBeatTime = configuration.getHeartBeatTime() * 1000;  
60 -  
61 63
62 - //向客户端发送消息  
63 - CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",  
64 - CIMRequestProto.CIMReqProtocol.class);  
65 - ctx.writeAndFlush(heartBeat).addListeners(ChannelFutureListener.CLOSE_ON_FAILURE); 64 + LOGGER.info("定时检测客户端端是否存活");
66 65
67 - Long lastReadTime = NettyAttrUtil.getReaderTime(ctx.channel());  
68 - long now = System.currentTimeMillis();  
69 - if (lastReadTime != null && now - lastReadTime > heartBeatTime){  
70 - CIMUserInfo userInfo = SessionSocketHolder.getUserId((NioSocketChannel) ctx.channel());  
71 - LOGGER.warn("客户端[{}]心跳超时[{}]ms，需要关闭连接!",userInfo.getUserName(),now - lastReadTime);  
72 - userOffLine(userInfo, (NioSocketChannel) ctx.channel());  
73 - ctx.channel().close();  
74 - } 66 + HeartBeatHandler heartBeatHandler = SpringBeanFactory.getBean(ServerHeartBeatHandlerImpl.class) ;
  67 + heartBeatHandler.process(ctx) ;
75 } 68 }
76 } 69 }
77 super.userEventTriggered(ctx, evt); 70 super.userEventTriggered(ctx, evt);
@@ -93,7 +86,7 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto @@ -93,7 +86,7 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
93 } 86 }
94 87
95 /** 88 /**
96 - * 清除路由关系 89 + * 下线，清除路由关系
97 * 90 *
98 * @param userInfo 91 * @param userInfo
99 * @throws IOException 92 * @throws IOException
@@ -137,6 +130,15 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto @@ -137,6 +130,15 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
137 //心跳更新时间 130 //心跳更新时间
138 if (msg.getType() == Constants.CommandType.PING){ 131 if (msg.getType() == Constants.CommandType.PING){
139 NettyAttrUtil.updateReaderTime(ctx.channel(),System.currentTimeMillis()); 132 NettyAttrUtil.updateReaderTime(ctx.channel(),System.currentTimeMillis());
  133 + //向客户端响应 pong 消息
  134 + CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",
  135 + CIMRequestProto.CIMReqProtocol.class);
  136 + ctx.writeAndFlush(heartBeat).addListeners((ChannelFutureListener) future -> {
  137 + if (!future.isSuccess()) {
  138 + LOGGER.error("IO error,close Channel");
  139 + future.channel().close();
  140 + }
  141 + }) ;
140 } 142 }
141 143
142 } 144 }
@@ -25,8 +25,8 @@ public class CIMServerInitializer extends ChannelInitializer<Channel> { @@ -25,8 +25,8 @@ public class CIMServerInitializer extends ChannelInitializer<Channel> {
25 protected void initChannel(Channel ch) throws Exception { 25 protected void initChannel(Channel ch) throws Exception {
26 26
27 ch.pipeline() 27 ch.pipeline()
28 - //30 秒没有向客户端发送消息就发生心跳  
29 - .addLast(new IdleStateHandler(30, 0, 0)) 28 + //11 秒没有向客户端发送消息就发生心跳
  29 + .addLast(new IdleStateHandler(11, 0, 0))
30 // google Protobuf 编解码 30 // google Protobuf 编解码
31 .addLast(new ProtobufVarint32FrameDecoder()) 31 .addLast(new ProtobufVarint32FrameDecoder())
32 .addLast(new ProtobufDecoder(CIMRequestProto.CIMReqProtocol.getDefaultInstance())) 32 .addLast(new ProtobufDecoder(CIMRequestProto.CIMReqProtocol.getDefaultInstance()))
  1 +package com.crossoverjie.cim.server.kit;
  2 +
  3 +import com.alibaba.fastjson.JSONObject;
  4 +import com.crossoverjie.cim.common.pojo.CIMUserInfo;
  5 +import com.crossoverjie.cim.server.config.AppConfiguration;
  6 +import com.crossoverjie.cim.server.util.SessionSocketHolder;
  7 +import com.crossoverjie.cim.server.util.SpringBeanFactory;
  8 +import io.netty.channel.socket.nio.NioSocketChannel;
  9 +import okhttp3.*;
  10 +import org.slf4j.Logger;
  11 +import org.slf4j.LoggerFactory;
  12 +import org.springframework.stereotype.Component;
  13 +
  14 +import java.io.IOException;
  15 +
  16 +/**
  17 + * Function:
  18 + *
  19 + * @author crossoverJie
  20 + * Date: 2019-01-20 17:20
  21 + * @since JDK 1.8
  22 + */
  23 +@Component
  24 +public class RouteHandler {
  25 + private final static Logger LOGGER = LoggerFactory.getLogger(RouteHandler.class);
  26 +
  27 +
  28 + private final MediaType mediaType = MediaType.parse("application/json");
  29 +
  30 + /**
  31 + * 用户下线
  32 + * @param userInfo
  33 + * @param channel
  34 + * @throws IOException
  35 + */
  36 + public void userOffLine(CIMUserInfo userInfo, NioSocketChannel channel) throws IOException {
  37 + if (userInfo != null){
  38 + LOGGER.info("用户[{}]下线", userInfo.getUserName());
  39 + SessionSocketHolder.removeSession(userInfo.getUserId());
  40 + //清除路由关系
  41 + clearRouteInfo(userInfo);
  42 + }
  43 + SessionSocketHolder.remove(channel);
  44 +
  45 + }
  46 +
  47 +
  48 + /**
  49 + * 清除路由关系
  50 + *
  51 + * @param userInfo
  52 + * @throws IOException
  53 + */
  54 + private void clearRouteInfo(CIMUserInfo userInfo) throws IOException {
  55 + OkHttpClient okHttpClient = SpringBeanFactory.getBean(OkHttpClient.class);
  56 + AppConfiguration configuration = SpringBeanFactory.getBean(AppConfiguration.class);
  57 + JSONObject jsonObject = new JSONObject();
  58 + jsonObject.put("userId", userInfo.getUserId());
  59 + jsonObject.put("msg", "offLine");
  60 + RequestBody requestBody = RequestBody.create(mediaType, jsonObject.toString());
  61 +
  62 + Request request = new Request.Builder()
  63 + .url(configuration.getClearRouteUrl())
  64 + .post(requestBody)
  65 + .build();
  66 +
  67 + Response response = null;
  68 + try {
  69 + response = okHttpClient.newCall(request).execute();
  70 + if (!response.isSuccessful()) {
  71 + throw new IOException("Unexpected code " + response);
  72 + }
  73 + } finally {
  74 + response.body().close();
  75 + }
  76 + }
  77 +
  78 +}
  1 +package com.crossoverjie.cim.server.kit;
  2 +
  3 +import com.crossoverjie.cim.common.kit.HeartBeatHandler;
  4 +import com.crossoverjie.cim.common.pojo.CIMUserInfo;
  5 +import com.crossoverjie.cim.common.util.NettyAttrUtil;
  6 +import com.crossoverjie.cim.server.config.AppConfiguration;
  7 +import com.crossoverjie.cim.server.util.SessionSocketHolder;
  8 +import io.netty.channel.ChannelHandlerContext;
  9 +import io.netty.channel.socket.nio.NioSocketChannel;
  10 +import org.slf4j.Logger;
  11 +import org.slf4j.LoggerFactory;
  12 +import org.springframework.beans.factory.annotation.Autowired;
  13 +import org.springframework.stereotype.Service;
  14 +
  15 +/**
  16 + * Function:
  17 + *
  18 + * @author crossoverJie
  19 + * Date: 2019-01-20 17:16
  20 + * @since JDK 1.8
  21 + */
  22 +@Service
  23 +public class ServerHeartBeatHandlerImpl implements HeartBeatHandler {
  24 +
  25 + private final static Logger LOGGER = LoggerFactory.getLogger(ServerHeartBeatHandlerImpl.class);
  26 +
  27 + @Autowired
  28 + private RouteHandler routeHandler ;
  29 +
  30 + @Autowired
  31 + private AppConfiguration appConfiguration ;
  32 +
  33 + @Override
  34 + public void process(ChannelHandlerContext ctx) throws Exception {
  35 +
  36 + long heartBeatTime = appConfiguration.getHeartBeatTime() * 1000;
  37 +
  38 + Long lastReadTime = NettyAttrUtil.getReaderTime(ctx.channel());
  39 + long now = System.currentTimeMillis();
  40 + if (lastReadTime != null && now - lastReadTime > heartBeatTime){
  41 + CIMUserInfo userInfo = SessionSocketHolder.getUserId((NioSocketChannel) ctx.channel());
  42 + if (userInfo != null){
  43 + LOGGER.warn("客户端[{}]心跳超时[{}]ms，需要关闭连接!",userInfo.getUserName(),now - lastReadTime);
  44 + }
  45 + routeHandler.userOffLine(userInfo, (NioSocketChannel) ctx.channel());
  46 + ctx.channel().close();
  47 + }
  48 + }
  49 +}
@@ -34,4 +34,4 @@ app.zk.root=/route @@ -34,4 +34,4 @@ app.zk.root=/route
34 cim.clear.route.request.url=http://localhost:8083/offLine 34 cim.clear.route.request.url=http://localhost:8083/offLine
35 35
36 # 检测多少秒没有收到客户端心跳后服务端关闭连接 36 # 检测多少秒没有收到客户端心跳后服务端关闭连接
37 -cim.heartbeat.time = 40  
  37 +cim.heartbeat.time = 30
@@ -30,7 +30,6 @@ @@ -30,7 +30,6 @@
30 <module>cim-server</module> 30 <module>cim-server</module>
31 <module>cim-client</module> 31 <module>cim-client</module>
32 <module>cim-common</module> 32 <module>cim-common</module>
33 - <module>springboot-admin</module>  
34 <module>cim-zk</module> 33 <module>cim-zk</module>
35 <module>cim-forward-route</module> 34 <module>cim-forward-route</module>
36 </modules> 35 </modules>
1 -# 监控 admin  
2 -  
3 -[SpringBoot admin](https://github.com/codecentric/spring-boot-admin)  
1 -<?xml version="1.0" encoding="UTF-8"?>  
2 -<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
3 - xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
4 - <modelVersion>4.0.0</modelVersion>  
5 -  
6 - <parent>  
7 - <groupId>com.crossoverjie.netty</groupId>  
8 - <artifactId>cim</artifactId>  
9 - <version>1.0.0-SNAPSHOT</version>  
10 - </parent>  
11 -  
12 - <artifactId>springboot-admin</artifactId>  
13 - <packaging>jar</packaging>  
14 -  
15 - <name>admin</name>  
16 - <description>springBoot admin</description>  
17 -  
18 -  
19 - <properties>  
20 - <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>  
21 - <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>  
22 - <java.version>1.8</java.version>  
23 - <logback.version>1.2.3</logback.version>  
24 - </properties>  
25 -  
26 - <dependencies>  
27 - <dependency>  
28 - <groupId>org.springframework.boot</groupId>  
29 - <artifactId>spring-boot-starter</artifactId>  
30 - </dependency>  
31 -  
32 - <dependency>  
33 - <groupId>org.springframework.boot</groupId>  
34 - <artifactId>spring-boot-starter-test</artifactId>  
35 - <scope>test</scope>  
36 - </dependency>  
37 -  
38 - <dependency>  
39 - <groupId>org.springframework.boot</groupId>  
40 - <artifactId>spring-boot-starter-mail</artifactId>  
41 - </dependency>  
42 -  
43 - <dependency>  
44 - <groupId>de.codecentric</groupId>  
45 - <artifactId>spring-boot-admin-starter-server</artifactId>  
46 - <version>1.5.7</version>  
47 - </dependency>  
48 -  
49 - <dependency>  
50 - <groupId>ch.qos.logback</groupId>  
51 - <artifactId>logback-classic</artifactId>  
52 - </dependency>  
53 -  
54 - <dependency>  
55 - <groupId>de.codecentric</groupId>  
56 - <artifactId>spring-boot-admin-server-ui</artifactId>  
57 - <version>1.5.6</version>  
58 - </dependency>  
59 - </dependencies>  
60 -  
61 - <build>  
62 - <plugins>  
63 - <plugin>  
64 - <groupId>org.springframework.boot</groupId>  
65 - <artifactId>spring-boot-maven-plugin</artifactId>  
66 - </plugin>  
67 - </plugins>  
68 - </build>  
69 -  
70 -  
71 -</project>  
1 -package com.ai.obc.springboot.admin;  
2 -  
3 -import de.codecentric.boot.admin.config.EnableAdminServer;  
4 -import org.springframework.boot.SpringApplication;  
5 -import org.springframework.boot.autoconfigure.EnableAutoConfiguration;  
6 -import org.springframework.boot.autoconfigure.SpringBootApplication;  
7 -import org.springframework.context.annotation.Configuration;  
8 -  
9 -@SpringBootApplication  
10 -@Configuration  
11 -@EnableAutoConfiguration  
12 -@EnableAdminServer  
13 -public class AdminApplication {  
14 -  
15 - public static void main(String[] args) {  
16 - SpringApplication.run(AdminApplication.class, args);  
17 - }  
18 -}  
1 -  
2 -spring.application.name=spring-boot-admin  
3 -server.cimServerPort = 8888  
4 -  
5 -logging.level.root=info  
1 -package com.ai.obc.springboot.admin;  
2 -  
3 -import org.junit.Test;  
4 -import org.junit.runner.RunWith;  
5 -import org.springframework.boot.test.context.SpringBootTest;  
6 -import org.springframework.test.context.junit4.SpringRunner;  
7 -  
8 -@RunWith(SpringRunner.class)  
9 -@SpringBootTest  
10 -public class AdminApplicationTests {  
11 -  
12 - @Test  
13 - public void contextLoads() {  
14 - }  
15 -  
16 -}