Merge pull request #26 from crossoverJie/cim-1.0.2

cim 1.0.2

Author: crossoverJie

cim-client/pom.xml

@@ -72,12 +72,6 @@
             <artifactId>logback-classic</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>io.netty</groupId>
-            <artifactId>netty-all</artifactId>
-            <version>${netty.version}</version>
-        </dependency>
-
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

cim-client/src/main/java/com/crossoverjie/cim/client/CIMClientApplication.java

@@ -1,8 +1,10 @@
 package com.crossoverjie.cim.client;
 
 import com.crossoverjie.cim.client.scanner.Scan;
+import com.crossoverjie.cim.client.service.impl.ClientInfo;
 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;
@@ -15,6 +17,8 @@ public class CIMClientApplication implements CommandLineRunner{
 
 	private final static Logger LOGGER = LoggerFactory.getLogger(CIMClientApplication.class);
 
+	@Autowired
+	private ClientInfo clientInfo ;
 	public static void main(String[] args) {
         SpringApplication.run(CIMClientApplication.class, args);
 		LOGGER.info("启动 Client 服务成功");
@@ -26,5 +30,6 @@ public void run(String... args) throws Exception {
 		Thread thread = new Thread(scan);
 		thread.setName("scan-thread");
 		thread.start();
+		clientInfo.saveStartDate();
 	}
 }

cim-client/src/main/java/com/crossoverjie/cim/client/client/CIMClient.java

@@ -1,7 +1,10 @@
 package com.crossoverjie.cim.client.client;
 
+import com.crossoverjie.cim.client.config.AppConfiguration;
 import com.crossoverjie.cim.client.init.CIMClientHandleInitializer;
+import com.crossoverjie.cim.client.service.MsgHandle;
 import com.crossoverjie.cim.client.service.RouteRequest;
+import com.crossoverjie.cim.client.service.impl.ClientInfo;
 import com.crossoverjie.cim.client.vo.req.GoogleProtocolVO;
 import com.crossoverjie.cim.client.vo.req.LoginReqVO;
 import com.crossoverjie.cim.client.vo.res.CIMServerResVO;
@@ -16,6 +19,7 @@
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.concurrent.DefaultThreadFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -28,15 +32,15 @@
  * Function:
  *
  * @author crossoverJie
- *         Date: 22/05/2018 14:19
+ * Date: 22/05/2018 14:19
  * @since JDK 1.8
  */
 @Component
 public class CIMClient {
 
     private final static Logger LOGGER = LoggerFactory.getLogger(CIMClient.class);
 
-    private EventLoopGroup group = new NioEventLoopGroup();
+    private EventLoopGroup group = new NioEventLoopGroup(0, new DefaultThreadFactory("cim-work"));
 
     @Value("${cim.user.id}")
     private long userId;
@@ -49,6 +53,20 @@ public class CIMClient {
     @Autowired
     private RouteRequest routeRequest;
 
+    @Autowired
+    private AppConfiguration configuration;
+
+    @Autowired
+    private MsgHandle msgHandle;
+
+    @Autowired
+    private ClientInfo clientInfo;
+
+    /**
+     * 重试次数
+     */
+    private int errorCount;
+
     @PostConstruct
     public void start() throws Exception {
 
@@ -70,14 +88,25 @@ public void start() throws Exception {
      * @param cimServer
      * @throws InterruptedException
      */
-    private void startClient(CIMServerResVO.ServerInfo cimServer) throws InterruptedException {
+    private void startClient(CIMServerResVO.ServerInfo cimServer) {
         Bootstrap bootstrap = new Bootstrap();
         bootstrap.group(group)
                 .channel(NioSocketChannel.class)
                 .handler(new CIMClientHandleInitializer())
         ;
 
-        ChannelFuture future = bootstrap.connect(cimServer.getIp(), cimServer.getCimServerPort()).sync();
+        ChannelFuture future = null;
+        try {
+            future = bootstrap.connect(cimServer.getIp(), cimServer.getCimServerPort()).sync();
+        } catch (InterruptedException e) {
+            errorCount++;
+
+            if (errorCount >= configuration.getErrorCount()) {
+                LOGGER.error("链接失败次数达到上限[{}]次", errorCount);
+                msgHandle.shutdown();
+            }
+            LOGGER.error("连接失败", e);
+        }
         if (future.isSuccess()) {
             LOGGER.info("启动 cim client 成功");
         }
@@ -90,10 +119,26 @@ private void startClient(CIMServerResVO.ServerInfo cimServer) throws Interrupted
      * @return 路由服务器信息
      * @throws Exception
      */
-    private CIMServerResVO.ServerInfo userLogin() throws Exception {
+    private CIMServerResVO.ServerInfo userLogin() {
         LoginReqVO loginReqVO = new LoginReqVO(userId, userName);
-        CIMServerResVO.ServerInfo cimServer = routeRequest.getCIMServer(loginReqVO);
-        LOGGER.info("cimServer=[{}]", cimServer.toString());
+        CIMServerResVO.ServerInfo cimServer = null;
+        try {
+            cimServer = routeRequest.getCIMServer(loginReqVO);
+
+            //保存系统信息
+            clientInfo.saveServiceInfo(cimServer.getIp() + ":" + cimServer.getCimServerPort())
+                    .saveUserInfo(userId, userName);
+
+            LOGGER.info("cimServer=[{}]", cimServer.toString());
+        } catch (Exception e) {
+            errorCount++;
+
+            if (errorCount >= configuration.getErrorCount()) {
+                LOGGER.error("重连次数达到上限[{}]次", errorCount);
+                msgHandle.shutdown();
+            }
+            LOGGER.error("登录失败", e);
+        }
         return cimServer;
     }
 
@@ -145,11 +190,25 @@ public void sendGoogleProtocolMsg(GoogleProtocolVO googleProtocolVO) {
 
     }
 
+
+    public void reconnect() throws Exception {
+        if (channel != null && channel.isActive()) {
+            return;
+        }
+        //首先清除路由信息，下线
+        routeRequest.offLine();
+
+        LOGGER.info("开始重连。。");
+        start();
+        LOGGER.info("重连成功！！");
+    }
+
     /**
      * 关闭
+     *
      * @throws InterruptedException
      */
     public void close() throws InterruptedException {
-        channel.close() ;
+        channel.close();
     }
 }

cim-client/src/main/java/com/crossoverjie/cim/client/config/AppConfiguration.java

@@ -22,6 +22,15 @@ public class AppConfiguration {
     @Value("${cim.msg.logger.path}")
     private String msgLoggerPath ;
 
+    @Value("${cim.clear.route.request.url}")
+    private String clearRouteUrl ;
+
+    @Value("${cim.heartbeat.time}")
+    private long heartBeatTime ;
+
+    @Value("${cim.reconnect.count}")
+    private int errorCount ;
+
     public Long getUserId() {
         return userId;
     }
@@ -45,4 +54,30 @@ public String getMsgLoggerPath() {
     public void setMsgLoggerPath(String msgLoggerPath) {
         this.msgLoggerPath = msgLoggerPath;
     }
+
+
+    public long getHeartBeatTime() {
+        return heartBeatTime;
+    }
+
+    public void setHeartBeatTime(long heartBeatTime) {
+        this.heartBeatTime = heartBeatTime;
+    }
+
+
+    public String getClearRouteUrl() {
+        return clearRouteUrl;
+    }
+
+    public void setClearRouteUrl(String clearRouteUrl) {
+        this.clearRouteUrl = clearRouteUrl;
+    }
+
+    public int getErrorCount() {
+        return errorCount;
+    }
+
+    public void setErrorCount(int errorCount) {
+        this.errorCount = errorCount;
+    }
 }

cim-client/src/main/java/com/crossoverjie/cim/client/config/BeanConfig.java

@@ -82,6 +82,17 @@ public ThreadPoolExecutor buildCallerThread(){
         return  productExecutor ;
     }
 
+
+    @Bean("scheduledTask")
+    public ScheduledExecutorService buildSchedule(){
+        ThreadFactory sche = new ThreadFactoryBuilder()
+                .setNameFormat("scheduled-%d")
+                .setDaemon(true)
+                .build();
+        ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1,sche) ;
+        return scheduledExecutorService ;
+    }
+
     /**
      * 回调 bean
      * @return

cim-client/src/main/java/com/crossoverjie/cim/client/handle/CIMClientHandle.java

@@ -1,9 +1,11 @@
 package com.crossoverjie.cim.client.handle;
 
+import com.crossoverjie.cim.client.thread.ReConnectJob;
 import com.crossoverjie.cim.client.util.SpringBeanFactory;
 import com.crossoverjie.cim.common.constant.Constants;
 import com.crossoverjie.cim.common.protocol.CIMRequestProto;
 import com.crossoverjie.cim.common.protocol.CIMResponseProto;
+import com.crossoverjie.cim.common.util.NettyAttrUtil;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -13,7 +15,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Function:
@@ -31,20 +35,28 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
 
     private ThreadPoolExecutor threadPoolExecutor ;
 
+    private ScheduledExecutorService scheduledExecutorService ;
+
 
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
 
         if (evt instanceof IdleStateEvent){
             IdleStateEvent idleStateEvent = (IdleStateEvent) evt ;
 
+            //LOGGER.info("定时检测服务端是否存活");
+
             if (idleStateEvent.state() == IdleState.WRITER_IDLE){
                 CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",
                         CIMRequestProto.CIMReqProtocol.class);
-                ctx.writeAndFlush(heartBeat).addListeners(ChannelFutureListener.CLOSE_ON_FAILURE) ;
+                ctx.writeAndFlush(heartBeat).addListeners((ChannelFutureListener) future -> {
+                    if (!future.isSuccess()) {
+                        LOGGER.error("IO error,close Channel");
+                        future.channel().close();
+                    }
+                }) ;
             }
 
-
         }
 
         super.userEventTriggered(ctx, evt);
@@ -58,10 +70,24 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
     }
 
     @Override
-    protected void channelRead0(ChannelHandlerContext channelHandlerContext, CIMResponseProto.CIMResProtocol msg) throws Exception {
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        LOGGER.info("客户端断开了，重新连接！");
 
-        //从服务端收到消息时被调用
-        //LOGGER.info("客户端收到消息={}",in.toString(CharsetUtil.UTF_8)) ;
+        if (scheduledExecutorService == null){
+            scheduledExecutorService = SpringBeanFactory.getBean("scheduledTask",ScheduledExecutorService.class) ;
+        }
+        // TODO: 2019-01-22 后期可以改为不用定时任务，连上后就关闭任务 节省性能。
+        scheduledExecutorService.scheduleAtFixedRate(new ReConnectJob(ctx),0,10, TimeUnit.SECONDS) ;
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, CIMResponseProto.CIMResProtocol msg) throws Exception {
+
+        //心跳更新时间
+        if (msg.getType() == Constants.CommandType.PING){
+            //LOGGER.info("收到服务端心跳！！！");
+            NettyAttrUtil.updateReaderTime(ctx.channel(),System.currentTimeMillis());
+        }
 
         if (msg.getType() != Constants.CommandType.PING) {
             //回调消息
@@ -70,6 +96,10 @@ protected void channelRead0(ChannelHandlerContext channelHandlerContext, CIMResp
             LOGGER.info(msg.getResMsg());
         }
 
+
+
+
+
     }
 
     /**

cim-client/src/main/java/com/crossoverjie/cim/client/init/CIMClientHandleInitializer.java

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

cim-client/src/main/java/com/crossoverjie/cim/client/service/MsgHandle.java

@@ -48,4 +48,10 @@ public interface MsgHandle {
      * @return 是否应当跳过当前消息（包含了":" 就需要跳过）
      */
     boolean innerCommand(String msg) ;
+
+
+    /**
+     * 关闭系统
+     */
+    void shutdown() ;
 }

cim-client/src/main/java/com/crossoverjie/cim/client/service/RouteRequest.java

@@ -41,10 +41,13 @@ public interface RouteRequest {
     CIMServerResVO.ServerInfo getCIMServer(LoginReqVO loginReqVO) throws Exception;
 
     /**
-     *
-     * @return 获取所有在线用户
+     * 获取所有在线用户
+     * @return
+     * @throws Exception
      */
     List<OnlineUsersResVO.DataBodyBean> onlineUsers()throws Exception ;
 
 
+    void offLine() ;
+
 }