使用Netty搭建TCP服务器
1.添加pom文件,引入Netty依赖<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.65.Final</version>
</dependency>
2.添加ChannelHandlerContextMap类，用来存储通道的标识和通道对象
在长连接中，我们可以给每个通道附带一些属性，也可以说是标识，当你拿到这个通道对象时，可以获取这些属性，不外这里的前提是，需要先拿到通道对象，所以，我们无法通过通道标识去拿到通道对象，这样非常的不友好，比方，我们需要获取某个用户的通道，然后向这个通道推送消息，这个时候我们就只能循环每个通道，然后判断当前循环的通道是不是对应用户的，加了这个类，可以让我们快速的找到对应用户的通道，这里使用ConcurrentHashMap类，ConcurrentHashMap类是线程安全的。
import io.netty.channel.Channel;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
public class ChannelHandlerContextMap {
private static ChannelHandlerContextMap instance;
private final Map<String, Channel> channelMap;
private ChannelHandlerContextMap() {
channelMap = new ConcurrentHashMap<>();
}
public static synchronized ChannelHandlerContextMap getInstance() {
if (instance == null) {
instance = new ChannelHandlerContextMap();
}
return instance;
}
public void put(String identifier, Channel ctx) {
channelMap.put(identifier, ctx);
}
public Channel get(String identifier) {
return channelMap.get(identifier);
}
public Channel remove(String identifier) {
return channelMap.remove(identifier);
}
public Boolean containsKey(String identifier) {
return channelMap.containsKey(identifier);
}
public Set<Map.Entry<String, Channel>> entrySet() {
return channelMap.entrySet();
}
public int size() {
return channelMap.size();
}
}
3.添加HeartbeatHandler类，用来处置惩罚通道在一定时间内，和服务器没有消息往来的处置惩罚方案
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
public class HeartbeatHandler extends IdleStateHandler {
public HeartbeatHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
super(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds);
}
@Override
protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
String deviceNameAttr = ctx.channel().attr("deviceName").get();
if(!"Server".equals(deviceNameAttr)){
//LogUtil.LOGI(HeartbeatHandler.class.getName(),"客户端超时类型："+evt.state()+";通道标识："+deviceNameAttr);
if (evt.state() == IdleState.READER_IDLE) {
// 一段时间内没有收到过消息
LogUtil.LOGI(HeartbeatHandler.class.getName(),"客户端读取超时，关闭通道:"+deviceNameAttr);
ctx.close();
}
// if (evt.state() == IdleState.WRITER_IDLE) {
// //一段时间内没有向外发送过消息
// LogUtil.LOGI(HeartbeatHandler.class.getName(),"客户端写入超时，关闭通道:"+deviceNameAttr);
// //ctx.close();
// }
// if (evt.state() == IdleState.ALL_IDLE) {
// //一段时间内既没收到过消息也没发送过消息
// LogUtil.LOGI(HeartbeatHandler.class.getName(),"客户端超时，关闭通道:"+deviceNameAttr);
// //ctx.close();
// }
}
}
}
当一定时间内，通道和服务器没有消息往来时，会进入channelIdle方法，这里的一定时间，后面的代码中会设置；在channelIdle方法中，我们可以通过ctx.channel()来获取触发条件的通道，通过evt.state()来获取触发的范例，触发范例一共有三种：
IdleState.READER_IDLE：服务端一段时间内，没有收到客户端的消息，触发
IdleState.WRITER_IDLE：服务端一段时间内，没有向客户端发送消息，触发
IdleState.ALL_IDLE：服务端一段时间内，既没有收到客户端的消息，也没有向客户端发送消息，触发
可以在这里做本身的逻辑，我这边是直接关闭通道，防止通道占用过多
3.添加NettyHandler类，处置惩罚服务器收到的消息
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.Attribute;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.Set;
@Component
public class NettyHandler extends ChannelInboundHandlerAdapter {
public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
@Override
public void channelActive(ChannelHandlerContext ctx) {
LogUtil.LOGI(NettyHandler.class.getName(),"客户端["+ctx.channel().remoteAddress()+"]已连接");
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
String deviceName = ctx.channel().attr("deviceName").get();
if(deviceName!=null && ChannelHandlerContextMap.getInstance().containsKey(deviceName)){
ChannelHandlerContextMap.getInstance().remove(deviceName);
LogUtil.LOGI(NettyHandler.class.getName(),"客户端["+deviceName+"]断开连接");
}else{
LogUtil.LOGI(NettyHandler.class.getName(),"客户端["+ctx.channel().remoteAddress()+"]断开连接");
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg){
LogUtil.LOGI(NettyHandler.class.getName(),"源消息:"+message);
}
public void sendMessgae(Channel ctx,String message){
ctx.writeAndFlush(message);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
public String buildReturnPackage(int type,Object content){
JSONObject jsonObject = new JSONObject();
jsonObject.put("type",type);
jsonObject.put("typeName",ConnectCodeUtil.MESSAGE_TYPE.get(type));
jsonObject.put("content",content);
return jsonObject+"##@##";
}
}
channelActive:当客户端连接上服务端，会触发这个方法
channelInactive：当客户端断开服务端，会触发这个方法
exceptionCaught：当出现非常时，会触发这个方法
channelRead：当收到消息，会触发这个方法
我们需要在channelInactive、channelRead这两个方法中，管理好我们第二步创建的Map集合
在channelInactive方法中，可以看到，我并不是直接在Map中操作remove，是需要先判断该通道是否包罗标识，存在标识我才remove，这和我们的业务逻辑干系，我们的业务逻辑存在客户端连上了服务端，不外还没有认证（就是认证时，会给通道打上标识，而且添加到Map中）的环境。这里根据你们的业务逻辑进行处置惩罚。
因为我这边的逻辑是，当存在同一个标识的通道出现时，我会将之前的通道给关闭，这个场景应该很常见，关于这个场景有一个标题，当这个场景出现，我会在channelRead方法中将Map中的通道覆盖，然后在channelInactive方法中移除Map中的通道数据，这样就会造成存在挤号时，Map中的数据是小于channelGroup中的数据的，这个你们根据本身的业务需求解决就好了
buildReturnPackage方法，是我构建发送消息的一个通用方法，消息最后的##@##是分隔符，后面会讲到。
4.添加NettyChannelInitializer类
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.CharsetUtil;
public class NettyChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel channel) throws Exception {
//channel.pipeline().addLast(new LineBasedFrameDecoder(1024));
String delimiter = "##@##";
// 对服务端返回的消息通过_$进行分隔，并且每次查找的最大大小为1024字节
channel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,
Unpooled.wrappedBuffer(delimiter.getBytes())));
// 将分隔之后的字节数据转换为字符串
channel.pipeline().addLast(new StringDecoder());
channel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
channel.pipeline().addLast("heartbeatHandler", new HeartbeatHandler(90, 90, 90)); // 添加心跳处理器
channel.pipeline().addLast(new NettyHandler());
//channel.pipeline().addLast(new LoggingHandler(LogLevel.INFO));
}
}
这个类就是将自界说的NettyHandler和HeartbeatHandler都添加进来，HeartbeatHandler的参数就是HeartbeatHandler类的构造方法的入参，分别是：
读取超时时间：IdleState.READER_IDLE
写入超时时间：IdleState.WRITER_IDLE
读写超时时间：IdleState.ALL_IDLE
单位秒
这里的##@##的界说分割符解码器，因为在Netty传输消息中，可能会存在粘包和拆包的标题，界说分隔符解码器就是，让Netty服务器将两个分隔符中心的文本，当做一条消息，从而解决粘包和拆包的标题。
解码器尚有多种范例：
LineBasedFrameDecoder：界说一条消息已\r\n结尾
FixedLengthFrameDecoder：界说一条消息的固定长度
DelimiterBasedFrameDecoder：自界说分隔符
5.添加配置文件
server.port=8473
netty.port = 5555
netty.url= 127.0.0.1
我这里是.properties文件
6.添加启动类
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
@Component
public class NettyServer {
// 开启服务端口
public static void start(InetSocketAddress address) {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap()
.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.localAddress(address)
.childHandler(new NettyChannelInitializer())
.option(ChannelOption.SO_BACKLOG, 102400)
.childOption(ChannelOption.SO_KEEPALIVE, true);
// 绑定端口，开始接收进来的连接
ChannelFuture future = bootstrap.bind(address).sync();
LogUtil.LOGI(NettyServer.class.getName(),"开启Netty服务端口：" + address.getPort());
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
然后在springboot的启动类，添加下面代码，就能启动了
InetSocketAddress address = new InetSocketAddress(URL, PORT);
server.start(address);
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]