Netty实现TCP和Websocket服务端
前言
我们日常使用 SpringBoot 已经可以轻松创建 HTTP 服务了，当然 SpringBoot 同样提供了基于注解的形式也能很方便实现 Websocket 服务。但是当我们想要更加自定义通信协议时，Java 届已经有统一主流的方式，使用 Netty ，一个韩国人开发的框架。 本文将介绍如何实现使用一个端口，同时监听多种网络协议。
1. 编码实现
我们使用的 HTTP 或 Websocket 协议(其实是 HTTP 协议的升级版)都是应用层协议，一般都默认使用 Json 序列化进行数据传输，Websocket 默认也支持二进制传输，使用者基本不必关心消息编解码。 但 TCP 属于传输层协议，需要自行实现消息编解码。
1.1 创建服务器
package com.mayee.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class NettyServer {
public void bind(int port) {
/**
* 配置服务端的NIO线程组
* NioEventLoopGroup 是用来处理I/O操作的Reactor线程组
* bossGroup：用来接收进来的连接，workerGroup：用来处理已经被接收的连接,进行socketChannel的网络读写，
* bossGroup接收到连接后就会把连接信息注册到workerGroup
* workerGroup的EventLoopGroup默认的线程数是CPU核数的二倍
*/
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
/**
* ServerBootstrap 是一个启动NIO服务的辅助启动类
*/
ServerBootstrap bootstrap = new ServerBootstrap()
/**
* 设置group，将bossGroup， workerGroup线程组传递到ServerBootstrap
*/
.group(bossGroup, workerGroup)
/**
* option是设置 bossGroup，childOption是设置workerGroup
* netty 默认数据包传输大小为1024字节, 设置它可以自动调整下一次缓冲区建立时分配的空间大小，避免内存的浪费 最小 初始化 最大 (根据生产环境实际情况来定)
* 使用对象池，重用缓冲区
*/
// .option(ChannelOption.SO_BACKLOG, 1024) // 设置TCP缓冲区
.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576))
.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576))
// .childOption(ChannelOption.TCP_NODELAY,true)
// .childOption(ChannelOption.SO_KEEPALIVE, true) // 两小时内没有数据的通信时,TCP会自动发送一个活动探测数据报文
/**
* ServerSocketChannel是以NIO的selector为基础进行实现的，用来接收新的连接，这里告诉Channel通过NioServerSocketChannel获取新的连接
*/
.channel(NioServerSocketChannel.class) //设置NIO的模式
.localAddress(port)
.handler(new LoggingHandler(LogLevel.INFO))
/**
* 设置 I/O处理类,主要用于网络I/O事件，记录日志，编码、解码消息
*/
.childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel channel) {
channel.pipeline().addLast(new ProtocolSelectorDecoder());
}
});
/**
* 绑定端口，同步等待成功
*/
ChannelFuture future = bootstrap.bind().sync();
log.info("服务已启动，正在监听: {}", future.channel().localAddress());
/**
* 等待服务器监听端口关闭
*/
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
/**
* 退出，释放线程池资源
*/
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}1.2 创建消息解码器
package com.mayee.server;
import com.mayee.server.tcp.TcpChannelInboundHandlerAdapter;
import com.mayee.server.ws.WsChannelInboundHandler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.CharsetUtil;
import java.nio.charset.StandardCharsets;
import java.util.List;
public class ProtocolSelectorDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
ChannelPipeline pipeline = channelHandlerContext.pipeline();
// 目前仅用 netty 实现了 tcp 和 ws 协议
if (isWebSocketUrl(byteBuf)) {
addWebSocketHandlers(pipeline);
} else {
addTCPProtocolHandlers(pipeline);
}
pipeline.remove(this);
}
private static final String WEBSOCKET_LINE_PREFIX = "GET /ws";
private static final String WEBSOCKET_PREFIX = "/ws";
private boolean isWebSocketUrl(ByteBuf byteBuf) {
if (byteBuf.readableBytes() < WEBSOCKET_LINE_PREFIX.length()) {
return false;
}
byteBuf.markReaderIndex();
byte[] content = new byte[WEBSOCKET_LINE_PREFIX.length()];
byteBuf.readBytes(content);
byteBuf.resetReaderIndex();
String s = new String(content, CharsetUtil.UTF_8);
return s.equals(WEBSOCKET_LINE_PREFIX);
}
private void addWebSocketHandlers(ChannelPipeline pipeline) {
//websocket协议本身是基于http协议的，所以这边也要使用http解编码器
pipeline.addLast(new HttpServerCodec());
//以块的方式来写的处理器
pipeline.addLast(new ChunkedWriteHandler());
pipeline.addLast(new HttpObjectAggregator(8192));
pipeline.addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10));
pipeline.addLast(new WsChannelInboundHandler());//自定义消息处理类
}
private void addTCPProtocolHandlers(ChannelPipeline pipeline) {
// ChannelOutboundHandler，依照逆序执行
pipeline.addLast("encoder", new StringEncoder(StandardCharsets.UTF_8));
// 属于ChannelInboundHandler，依照顺序执行
pipeline.addLast("decoder", new StringDecoder(StandardCharsets.UTF_8));
// 自定义ChannelInboundHandlerAdapter
pipeline.addLast(new TcpChannelInboundHandlerAdapter());
}
}1.3 创建 Websocket 处理器
package com.mayee.server.ws;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class WsChannelInboundHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.info("与客户端建立连接，通道开启！");
//添加到channelGroup通道组
CoordinationChannelHandlerPool.channelGroup.add(ctx.channel());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
log.info("与客户端断开连接，通道关闭！");
//从channelGroup通道组删除
CoordinationChannelHandlerPool.channelGroup.remove(ctx.channel());
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
//接收的消息
log.info("收到客户端{}的数据：{}",ctx.channel().id(), msg.text());
// 单独发消息
sendMessage(ctx);
// 群发消息
// sendAllMessage();
}
private void sendMessage(ChannelHandlerContext ctx) throws InterruptedException {
String message = "Websocket response: hello";
ctx.writeAndFlush(new TextWebSocketFrame(message));
}
private void sendAllMessage() {
String message = "我是服务器，这是群发消息";
CoordinationChannelHandlerPool.channelGroup.writeAndFlush(new TextWebSocketFrame(message));
}
}1.4 创建 TCP 处理器
package com.mayee.server.tcp;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
/**
* @link <a href="https://blog.csdn.net/qq_30665009/article/details/125338469">Netty搭建基于TCP协议的服务端</a>
*/
@Slf4j
public class TcpChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter {
/**
* 从客户端收到新的数据时，这个方法会在收到消息时被调用
*
* @param ctx
* @param msg
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException {
log.info("channelRead:read msg: {}", msg.toString());
//回应客户端
ctx.write("tcp response: hello");
}
/**
* 从客户端收到新的数据、读取完成时调用
*
* @param ctx
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
log.info("channelReadComplete");
ctx.flush();
}
/**
* 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
*
* @param ctx
* @param cause
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException {
log.info("exceptionCaught");
cause.printStackTrace();
ctx.close();//抛出异常，断开与客户端的连接
}
/**
* 客户端与服务端第一次建立连接时 执行
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelActive(ctx);
ctx.channel().read();
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
//此处不能使用ctx.close()，否则客户端始终无法与服务端建立连接
log.info("channelActive: {}", clientIp + ctx.name());
}
/**
* 客户端与服务端 断连时 执行
*
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelInactive(ctx);
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
ctx.close(); //断开连接时，必须关闭，否则造成资源浪费，并发量很大情况下可能造成宕机
log.info("channelInactive: {}", clientIp);
}
/**
* 服务端当read超时, 会调用这个方法
*
* @param ctx
* @param evt
* @throws Exception
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException {
super.userEventTriggered(ctx, evt);
InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = insocket.getAddress().getHostAddress();
ctx.close();//超时时断开连接
log.info("userEventTriggered: {}", clientIp);
}
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
log.info("channelRegistered");
}
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
log.info("channelUnregistered");
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
log.info("channelWritabilityChanged");
}
}2. 服务响应
连接到 Websocket 服务： ws://127.0.0.1:9600/ws。 发送任意内容，服务端响应:
Websocket response: hello连接到 TCP 服务： ws://127.0.0.1:9600 发送任意内容，服务端响应:
tcp response: hello可以看到实现了 Websocket 和 TCP 服务运行在同一端口。 但是，有一说一，这写起来还是挺复杂的，网络协议这块 GO 原生支持已经很好了，实现起来也非常方便。
Tip：本文完整示例代码已上传至 GitHub
版权所有
版权归属：Mayee