作者 crossoverJie

:art: Improving structure / format of the code.

... ... @@ -73,12 +73,6 @@
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>${netty.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
... ...
... ... @@ -4,6 +4,7 @@ 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;
... ... @@ -41,7 +42,12 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
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();
}
}) ;
}
... ... @@ -58,10 +64,12 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
}
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, CIMResponseProto.CIMResProtocol msg) throws Exception {
protected void channelRead0(ChannelHandlerContext ctx, CIMResponseProto.CIMResProtocol msg) throws Exception {
//从服务端收到消息时被调用
//LOGGER.info("客户端收到消息={}",in.toString(CharsetUtil.UTF_8)) ;
//心跳更新时间
if (msg.getType() == Constants.CommandType.PING){
NettyAttrUtil.updateReaderTime(ctx.channel(),System.currentTimeMillis());
}
if (msg.getType() != Constants.CommandType.PING) {
//回调消息
... ...
package com.crossoverjie.cim.client.service.impl;
import com.crossoverjie.cim.common.kit.HeartBeatHandler;
import io.netty.channel.ChannelHandlerContext;
/**
* Function:
*
* @author crossoverJie
* Date: 2019-01-20 17:16
* @since JDK 1.8
*/
public class ClientHeartBeatHandlerImpl implements HeartBeatHandler {
@Override
public void process(ChannelHandlerContext ctx) throws Exception {
}
}
... ...
... ... @@ -51,4 +51,7 @@ cim.callback.thread.pool.size = 2
# 关闭健康检查权限
management.security.enabled=false
# SpringAdmin 地址
spring.boot.admin.url=http://127.0.0.1:8888
\ No newline at end of file
spring.boot.admin.url=http://127.0.0.1:8888
# 检测多少秒没有收到服务端端心跳后重新登录获取连接
cim.heartbeat.time = 40
\ No newline at end of file
... ...
... ... @@ -38,5 +38,11 @@
<groupId>com.github.sgroschupf</groupId>
<artifactId>zkclient</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>${netty.version}</version>
</dependency>
</dependencies>
</project>
\ No newline at end of file
... ...
package com.crossoverjie.cim.common.kit;
import io.netty.channel.ChannelHandlerContext;
/**
* Function:
*
* @author crossoverJie
* Date: 2019-01-20 17:15
* @since JDK 1.8
*/
public interface HeartBeatHandler {
/**
* 处理
*/
void process(ChannelHandlerContext ctx) throws Exception ;
}
... ...
package com.crossoverjie.cim.server.util;
package com.crossoverjie.cim.common.util;
import io.netty.channel.Channel;
import io.netty.util.Attribute;
... ...
... ... @@ -64,13 +64,6 @@
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>${netty.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
... ...
... ... @@ -3,13 +3,18 @@ package com.crossoverjie.cim.server.handle;
import com.alibaba.fastjson.JSONObject;
import com.crossoverjie.cim.common.constant.Constants;
import com.crossoverjie.cim.common.exception.CIMException;
import com.crossoverjie.cim.common.kit.HeartBeatHandler;
import com.crossoverjie.cim.common.pojo.CIMUserInfo;
import com.crossoverjie.cim.common.protocol.CIMRequestProto;
import com.crossoverjie.cim.common.util.NettyAttrUtil;
import com.crossoverjie.cim.server.config.AppConfiguration;
import com.crossoverjie.cim.server.util.NettyAttrUtil;
import com.crossoverjie.cim.server.kit.ServerHeartBeatHandlerImpl;
import com.crossoverjie.cim.server.util.SessionSocketHolder;
import com.crossoverjie.cim.server.util.SpringBeanFactory;
import io.netty.channel.*;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
... ... @@ -55,23 +60,19 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
if (evt instanceof IdleStateEvent) {
IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
if (idleStateEvent.state() == IdleState.READER_IDLE) {
AppConfiguration configuration = SpringBeanFactory.getBean(AppConfiguration.class);
long heartBeatTime = configuration.getHeartBeatTime() * 1000;
//向客户端发送消息
CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",
CIMRequestProto.CIMReqProtocol.class);
ctx.writeAndFlush(heartBeat).addListeners(ChannelFutureListener.CLOSE_ON_FAILURE);
Long lastReadTime = NettyAttrUtil.getReaderTime(ctx.channel());
long now = System.currentTimeMillis();
if (lastReadTime != null && now - lastReadTime > heartBeatTime){
CIMUserInfo userInfo = SessionSocketHolder.getUserId((NioSocketChannel) ctx.channel());
LOGGER.warn("客户端[{}]心跳超时[{}]ms,需要关闭连接!",userInfo.getUserName(),now - lastReadTime);
userOffLine(userInfo, (NioSocketChannel) ctx.channel());
ctx.channel().close();
}
ctx.writeAndFlush(heartBeat).addListeners((ChannelFutureListener) future -> {
if (!future.isSuccess()) {
LOGGER.error("IO error,close Channel");
future.channel().close();
}
}) ;
HeartBeatHandler heartBeatHandler = SpringBeanFactory.getBean(ServerHeartBeatHandlerImpl.class) ;
heartBeatHandler.process(ctx) ;
}
}
super.userEventTriggered(ctx, evt);
... ...
package com.crossoverjie.cim.server.kit;
import com.alibaba.fastjson.JSONObject;
import com.crossoverjie.cim.common.pojo.CIMUserInfo;
import com.crossoverjie.cim.server.config.AppConfiguration;
import com.crossoverjie.cim.server.util.SessionSocketHolder;
import com.crossoverjie.cim.server.util.SpringBeanFactory;
import io.netty.channel.socket.nio.NioSocketChannel;
import okhttp3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
* Function:
*
* @author crossoverJie
* Date: 2019-01-20 17:20
* @since JDK 1.8
*/
@Component
public class RouteHandler {
private final static Logger LOGGER = LoggerFactory.getLogger(RouteHandler.class);
private final MediaType mediaType = MediaType.parse("application/json");
/**
* 用户下线
* @param userInfo
* @param channel
* @throws IOException
*/
public void userOffLine(CIMUserInfo userInfo, NioSocketChannel channel) throws IOException {
LOGGER.info("用户[{}]下线", userInfo.getUserName());
SessionSocketHolder.remove(channel);
SessionSocketHolder.removeSession(userInfo.getUserId());
//清除路由关系
clearRouteInfo(userInfo);
}
/**
* 清除路由关系
*
* @param userInfo
* @throws IOException
*/
private void clearRouteInfo(CIMUserInfo userInfo) throws IOException {
OkHttpClient okHttpClient = SpringBeanFactory.getBean(OkHttpClient.class);
AppConfiguration configuration = SpringBeanFactory.getBean(AppConfiguration.class);
JSONObject jsonObject = new JSONObject();
jsonObject.put("userId", userInfo.getUserId());
jsonObject.put("msg", "offLine");
RequestBody requestBody = RequestBody.create(mediaType, jsonObject.toString());
Request request = new Request.Builder()
.url(configuration.getClearRouteUrl())
.post(requestBody)
.build();
Response response = null;
try {
response = okHttpClient.newCall(request).execute();
if (!response.isSuccessful()) {
throw new IOException("Unexpected code " + response);
}
} finally {
response.body().close();
}
}
}
... ...
package com.crossoverjie.cim.server.kit;
import com.crossoverjie.cim.common.kit.HeartBeatHandler;
import com.crossoverjie.cim.common.pojo.CIMUserInfo;
import com.crossoverjie.cim.common.util.NettyAttrUtil;
import com.crossoverjie.cim.server.config.AppConfiguration;
import com.crossoverjie.cim.server.util.SessionSocketHolder;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* Function:
*
* @author crossoverJie
* Date: 2019-01-20 17:16
* @since JDK 1.8
*/
@Service
public class ServerHeartBeatHandlerImpl implements HeartBeatHandler {
private final static Logger LOGGER = LoggerFactory.getLogger(ServerHeartBeatHandlerImpl.class);
@Autowired
private RouteHandler routeHandler ;
@Autowired
private AppConfiguration appConfiguration ;
@Override
public void process(ChannelHandlerContext ctx) throws Exception {
long heartBeatTime = appConfiguration.getHeartBeatTime() * 1000;
Long lastReadTime = NettyAttrUtil.getReaderTime(ctx.channel());
long now = System.currentTimeMillis();
if (lastReadTime != null && now - lastReadTime > heartBeatTime){
CIMUserInfo userInfo = SessionSocketHolder.getUserId((NioSocketChannel) ctx.channel());
LOGGER.warn("客户端[{}]心跳超时[{}]ms,需要关闭连接!",userInfo.getUserName(),now - lastReadTime);
routeHandler.userOffLine(userInfo, (NioSocketChannel) ctx.channel());
ctx.channel().close();
}
}
}
... ...