一、粘包与半包
粘包
现象：第一次发送abc，第二次发送def，接收abcdef。
原因：1、应用层，接收方 ByteBuf 设置太大（Netty 默认 1024）
2、传输层，滑动窗口，假设发送方 256 bytes 表示一个完整报文，但由于接收方处理不及时且窗口大小足够大（大于256 bytes），这 256 bytes 字节就会缓冲在接收方的滑动窗口中，当滑动窗口中缓冲了多个报文就会粘包。
3、Nagle算法会导致粘包的产生。
半包
现象：发送 abcdef，分两次接收abc和def
原因：1、应用层，接收方 ByteBuf 小于实际发送数据量。
2、传输层，滑动窗口，假设接收方的窗口只剩了 128 bytes，发送方的报文大小是 256 bytes，这时接收方窗口中无法容纳发送方的全部报文，发送方只能先发送前 128 bytes，等待 ack 后才能发送剩余部分，这就造成了半包。
3、数据链路层，MSS（最大报文段长度，用于在TCP连接建立时，收发双方协商通信时每一个报文段所能承载的最大数据长度）限制，当发送的数据超过 MSS 限制后，会将数据切分发送，就会造成半包。
解决方案
1、短链接
客户端每次向服务器发送数据以后，就与服务器断开连接，此时的消息边界为连接建立到连接断开。这时便无需使用滑动窗口等技术来缓冲数据，则不会发生粘包现象。但如果一次性数据发送过多，接收方无法一次性容纳所有数据，还是会发生半包现象，所以短链接无法解决半包现象。客户端代码如下：
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.debug("sending...");
ByteBuf buffer = ctx.alloc().buffer(16);
buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
ctx.writeAndFlush(buffer);
// 使用短链接，每次发送完毕后就断开连接
ctx.channel().close();
}
2、定长解码器
客户端于服务器约定一个最大长度，保证客户端每次发送的数据长度都不会大于该长度。若发送数据长度不足则需要补齐至该长度，服务器接收数据时，将接收到的数据按照约定的最大长度进行拆分，即使发送过程中产生了粘包，也可以通过定长解码器将数据正确地进行拆分。服务端可以使用FixedLengthFrameDecoder对数据进行定长解码。服务端代码如下
ch.pipeline().addLast(new FixedLengthFrameDecoder(16));
3、行解码器
行解码器的是通过分隔符对数据进行拆分来解决粘包半包问题的，可以通过LineBasedFrameDecoder(int maxLength)来拆分以换行符( )为分隔符的数据，也可以通过DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf... delimiters)来指定通过什么分隔符来拆分数据（可以传入多个分隔符）两种解码器都需要传入数据的最大长度，若超出最大长度，会抛出TooLongFrameException异常。服务端代码如下：
// 通过行解码器对粘包数据进行拆分，以
为分隔符
// 需要指定最大长度
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(64));
如果是以自定义分隔符 c 为分隔符
// 将分隔符放入ByteBuf中
ByteBuf bufSet = ch.alloc().buffer().writeBytes("\c".getBytes(StandardCharsets.UTF_8));
// 通过行解码器对粘包数据进行拆分，以 c 为分隔符
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(64, ch.alloc().buffer().writeBytes(bufSet)));
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
4、长度字段解码器
在传送数据时可以在数据中添加一个用于表示有用数据长度的字段，在解码时读取出这个用于表明长度的字段，同时读取其他相关参数，即可知道最终需要的数据是什么样子的。
LengthFieldBasedFrameDecoder解码器可以提供更为丰富的拆分方法，其构造方法有五个参数：
maxFrameLength 数据最大长度（包括附加信息、长度标识等内容）
lengthFieldOffset 数据长度标识的起始偏移量，用于指明数据第几个字节开始是用于标识有用字节长度的，因为前面可能还有其他附加信息
lengthFieldLength 数据长度标识所占字节数（用于指明有用数据的长度），数据中用于表示有用数据长度的标识所占的字节数
lengthAdjustment 长度表示与有用数据的偏移量，用于指明数据长度标识和有用数据之间的距离，因为两者之间还可能有附加信息
initialBytesToStrip 数据读取起点，不读取 0 ~ initialBytesToStrip 之间的数据
二、协议设计与解析
TCP/IP 中消息传输基于流的方式，没有边界，那么就需要协议划定消息的边界，指定通信双方要共同遵守的通信规则，例如：HTTP协议。
HTTP协议在请求行请求头中都有很多的内容，自己实现较为困难，可以使用HttpServerCodec作为服务器端的解码器与编码器，来处理HTTP请求。
// HttpServerCodec 中既有请求的解码器 HttpRequestDecoder 又有响应的编码器 HttpResponseEncoder
// Codec(CodeCombine) 一般代表该类既作为 编码器 又作为 解码器
public final class HttpServerCodec extends CombinedChannelDuplexHandler<HttpRequestDecoder, HttpResponseEncoder>
implements HttpServerUpgradeHandler.SourceCodec
服务端代码
public class HttpServer {
static final Logger log = LoggerFactory.getLogger(StudyServer.class);
public static void main(String[] args) {
NioEventLoopGroup group = new NioEventLoopGroup();
new ServerBootstrap()
.group(group)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
// 作为服务器，使用 HttpServerCodec 作为编码器与解码器
ch.pipeline().addLast(new HttpServerCodec());
// 服务器只处理HTTPRequest
ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) {
// 获得请求uri
log.debug(msg.uri());
// 获得完整响应，设置版本号与状态码
DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
// 设置响应内容
byte[] bytes = "<h1>Hello, World!</h1>".getBytes(StandardCharsets.UTF_8);
// 设置响应体长度，避免浏览器一直接收响应内容
response.headers().setInt(CONTENT_LENGTH, bytes.length);
// 设置响应体
response.content().writeBytes(bytes);
// 写回响应
ctx.writeAndFlush(response);
}
});
}
})
.bind(8080);
}
}
自定义协议
组成要素
魔数：用来在第一时间判定接收的数据是否为无效数据包
版本号：可以支持协议的升级
序列化算法：消息正文到底采用哪种序列化反序列化方式，如：json、protobuf、hessian、jdk
指令类型：是登录、注册、单聊、群聊… 跟业务相关
请求序号：为了双工通信，提供异步能力
正文长度
消息正文
1、定义消息体
@Data public abstract class Message implements Serializable { /** * 根据消息类型字节，获得对应的消息 class * @param messageType 消息类型字节 * @return 消息 class */ public static Class<? extends Message> getMessageClass(int messageType) { return messageClasses.get(messageType); } /** * 序列id */ private int sequenceId; /** * 消息类型 */ private int messageType; public abstract int getMessageType(); public static final int LoginRequestMessage = 0; public static final int LoginResponseMessage = 1; public static final int ChatRequestMessage = 2; public static final int ChatResponseMessage = 3; public static final int GroupCreateRequestMessage = 4; public static final int GroupCreateResponseMessage = 5; public static final int GroupJoinRequestMessage = 6; public static final int GroupJoinResponseMessage = 7; public static final int GroupQuitRequestMessage = 8; public static final int GroupQuitResponseMessage = 9; public static final int GroupChatRequestMessage = 10; public static final int GroupChatResponseMessage = 11; public static final int GroupMembersRequestMessage = 12; public static final int GroupMembersResponseMessage = 13; public static final int PingMessage = 14; public static final int PongMessage = 15; /** * 请求类型 byte 值 */ public static final int RPC_MESSAGE_TYPE_REQUEST = 101; /** * 响应类型 byte 值 */ public static final int RPC_MESSAGE_TYPE_RESPONSE = 102; private static final Map<Integer, Class<? extends Message>> messageClasses = new HashMap<>(); static { messageClasses.put(LoginRequestMessage, LoginRequestMessage.class); messageClasses.put(LoginResponseMessage, LoginResponseMessage.class); messageClasses.put(ChatRequestMessage, ChatRequestMessage.class); messageClasses.put(ChatResponseMessage, ChatResponseMessage.class); messageClasses.put(GroupCreateRequestMessage, GroupCreateRequestMessage.class); messageClasses.put(GroupCreateResponseMessage, GroupCreateResponseMessage.class); messageClasses.put(GroupJoinRequestMessage, GroupJoinRequestMessage.class); messageClasses.put(GroupJoinResponseMessage, GroupJoinResponseMessage.class); messageClasses.put(GroupQuitRequestMessage, GroupQuitRequestMessage.class); messageClasses.put(GroupQuitResponseMessage, GroupQuitResponseMessage.class); messageClasses.put(GroupChatRequestMessage, GroupChatRequestMessage.class); messageClasses.put(GroupChatResponseMessage, GroupChatResponseMessage.class); messageClasses.put(GroupMembersRequestMessage, GroupMembersRequestMessage.class); messageClasses.put(GroupMembersResponseMessage, GroupMembersResponseMessage.class); messageClasses.put(RPC_MESSAGE_TYPE_REQUEST, RpcRequestMessage.class); messageClasses.put(RPC_MESSAGE_TYPE_RESPONSE, RpcResponseMessage.class); } }
2、编码器与解码器
@ChannelHandler.Sharable public class MessageCodec extends ByteToMessageCodec<Message> { @Override public void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception { // 1. 4 字节的魔数 out.writeBytes(new byte[]{1, 2, 3, 4}); // 2. 1 字节的版本, out.writeByte(1); // 3. 1 字节的序列化方式 jdk 0 , json 1 out.writeByte(0); // 4. 1 字节的指令类型 out.writeByte(msg.getMessageType()); // 5. 4 个字节 out.writeInt(msg.getSequenceId()); // 无意义，对齐填充 out.writeByte(0xff); // 6. 获取内容的字节数组 ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutputStream oos = new ObjectOutputStream(bos); oos.writeObject(msg); byte[] bytes = bos.toByteArray(); // 7. 长度 out.writeInt(bytes.length); // 8. 写入内容 out.writeBytes(bytes); } @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { // 获取魔数 int magicNum = in.readInt(); // 获取版本号 byte version = in.readByte(); // 获得序列化方式 byte serializerType = in.readByte(); // 获得指令类型 byte messageType = in.readByte(); // 获得请求序号 int sequenceId = in.readInt(); // 移除补齐字节 in.readByte(); // 获得正文长度 int length = in.readInt(); // 获得正文 byte[] bytes = new byte[length]; in.readBytes(bytes, 0, length); ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes)); Message message = (Message) ois.readObject(); log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length); log.debug("{}", message); // 将信息放入List中，传递给下一个handler out.add(message); } }
编码器与解码器方法源于父类ByteToMessageCodec，通过该类可以自定义编码器与解码器，泛型类型为被编码与被解码的类。此处使用了自定义类Message，代表消息。
编码器负责将附加信息与正文信息写入到ByteBuf中，其中附加信息总字节数最好为2^n，不足需要补齐。正文内容如果为对象，需要通过序列化将其放入到ByteBuf中。
解码器负责将ByteBuf中的信息取出，并放入List中，该List用于将信息传递给下一个handler。
编写测试类
public class TestCodec { public static void main(String[] args) throws Exception { EmbeddedChannel channel = new EmbeddedChannel(); // 添加解码器，避免粘包半包问题 channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 12, 4, 0, 0)); channel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG)); channel.pipeline().addLast(new MessageCodec()); LoginRequestMessage user = new LoginRequestMessage("Nyima", "123"); // 测试编码与解码 ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer(); new MessageCodec().encode(null, user, byteBuf); channel.writeInbound(byteBuf); } }
测试类中用到了LengthFieldBasedFrameDecoder，避免粘包半包问题。
通过MessageCodec的encode方法将附加信息与正文写入到ByteBuf中，通过channel执行入站操作，入站时会调用decode方法进行解码。
3、@ChannelHandler.Sharable
为了提高handler的复用率，可以将handler创建为handler对象，然后在不同的channel中使用该handler对象进行处理操作。但是并不是所有的handler都能通过这种方法来提高复用率的，例如：LengthFieldBasedFrameDecoder。如果多个channel中使用同一个LengthFieldBasedFrameDecoder对象，则可能发生如下问题：
1）channel1中收到了一个半包，LengthFieldBasedFrameDecoder发现不是一条完整的数据，则没有继续向下传播。
2）此时channel2中也收到了一个半包，因为两个channel使用了同一个LengthFieldBasedFrameDecoder，存入其中的数据刚好拼凑成了一个完整的数据包。LengthFieldBasedFrameDecoder让该数据包继续向下传播，最终引发错误。
为了提高handler的复用率，同时又避免出现一些并发问题，Netty中原生的handler中用@Sharable注解来标明，该handler能否在多个channel中共享。加了注解可以被共享，不加则不能被共享。
那么，我们自定义编解码器能否使用@Sharable注解，这需要根据自定义的handler的处理逻辑进行分析。我们的MessageCodec本身接收的是LengthFieldBasedFrameDecoder处理之后的数据，那么数据肯定是完整的，按分析来说是可以添加@Sharable注解的，
但是实际情况我们并不能添加该注解，会抛出异常信息ChannelHandler cn.nyimac.study.day8.protocol.MessageCodec is not allowed to be shared，因为MessageCodec继承自ByteToMessageCodec，ByteToMessageCodec类的注解如下：
这就意味着ByteToMessageCodec不能被多个channel所共享的。
原因：因为该类的目标是将ByteBuf转化为Message，意味着传进该handler的数据还未被处理过。所以传过来的ByteBuf可能并不是完整的数据，如果共享则会出现问题。
如果想要加注解共享怎么办呢？
继承MessageToMessageDecoder即可。该类的目标是：将已经被处理的完整数据再次被处理。传过来的Message如果是被处理过的完整数据，那么被共享也就不会出现问题了，也就可以使用@Sharable注解了。
实现方式与ByteToMessageCodec类似：
@ChannelHandler.Sharable public class MessageSharableCodec extends MessageToMessageCodec<ByteBuf, Message> { @Override protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> out) throws Exception { ... } @Override protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception { ... } }
三、案例应用——在线聊天室
代码已上传至码云：https://gitee.com/sglx666/netty-demo.git
1、用户登录接口
public interface UserService { /** * 登录 * @param username 用户名 * @param password 密码 * @return 登录成功返回 true, 否则返回 false */ boolean login(String username, String password); }
2、用户会话接口
public interface Session { /** * 绑定会话 * @param channel 哪个 channel 要绑定会话 * @param username 会话绑定用户 */ void bind(Channel channel, String username); /** * 解绑会话 * @param channel 哪个 channel 要解绑会话 */ void unbind(Channel channel); /** * 获取属性 * @param channel 哪个 channel * @param name 属性名 * @return 属性值 */ Object getAttribute(Channel channel, String name); /** * 设置属性 * @param channel 哪个 channel * @param name 属性名 * @param value 属性值 */ void setAttribute(Channel channel, String name, Object value); /** * 根据用户名获取 channel * @param username 用户名 * @return channel */ Channel getChannel(String username); }
3、群聊会话接口
public interface GroupSession { /** * 创建一个聊天组, 如果不存在才能创建成功, 否则返回 null * @param name 组名 * @param members 成员 * @return 成功时返回组对象, 失败返回 null */ Group createGroup(String name, Set<String> members); /** * 加入聊天组 * @param name 组名 * @param member 成员名 * @return 如果组不存在返回 null, 否则返回组对象 */ Group joinMember(String name, String member); /** * 移除组成员 * @param name 组名 * @param member 成员名 * @return 如果组不存在返回 null, 否则返回组对象 */ Group removeMember(String name, String member); /** * 移除聊天组 * @param name 组名 * @return 如果组不存在返回 null, 否则返回组对象 */ Group removeGroup(String name); /** * 获取组成员 * @param name 组名 * @return 成员集合, 如果群不存在或没有成员会返回 empty set */ Set<String> getMembers(String name); /** * 获取组成员的 channel 集合, 只有在线的 channel 才会返回 * @param name 组名 * @return 成员 channel 集合 */ List<Channel> getMembersChannel(String name); /** * 判断群聊是否一被创建 * @param name 群聊名称 * @return 是否存在 */ boolean isCreated(String name); }
4、客户端启动
public class ChatClient { static final Logger log = LoggerFactory.getLogger(ChatClient.class); public static void main(String[] args) { NioEventLoopGroup group = new NioEventLoopGroup(); LoggingHandler loggingHandler = new LoggingHandler(LogLevel.DEBUG); MessageSharableCodec messageSharableCodec = new MessageSharableCodec(); try { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group); bootstrap.channel(NioSocketChannel.class); bootstrap.handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new ProtocolFrameDecoder()); ch.pipeline().addLast(loggingHandler); ch.pipeline().addLast(messageSharableCodec); } }); Channel channel = bootstrap.connect().sync().channel(); channel.closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); } finally { group.shutdownGracefully(); } } }
5、服务端启动
public class ChatServer { static final Logger log = LoggerFactory.getLogger(ChatServer.class); public static void main(String[] args) { NioEventLoopGroup boss = new NioEventLoopGroup(); NioEventLoopGroup worker = new NioEventLoopGroup(); LoggingHandler loggingHandler = new LoggingHandler(LogLevel.DEBUG); MessageSharableCodec messageSharableCodec = new MessageSharableCodec(); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(boss, worker); bootstrap.channel(NioServerSocketChannel.class); bootstrap.childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new ProtocolFrameDecoder()); ch.pipeline().addLast(loggingHandler); ch.pipeline().addLast(messageSharableCodec); } }); Channel channel = bootstrap.bind(8080).sync().channel(); channel.closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); } finally { boss.shutdownGracefully(); worker.shutdownGracefully(); } } }
6、增加逻辑处理
客户端登录
向socketchannel中添加inbound handler，
//处理用户登录输入用户名密码的校验，以及后续的操作 ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){ ··· });
除了直接new ChannelInboundHandlerAdapter()，也可以自定义一个类，例如：MessageChannelInboundHandler继承ChannelInboundHandlerAdapter，并重写其中的方法。
服务端登录
自定义handler，负责处理登录请求信息，并作出响应
@ChannelHandler.Sharable // 必须添加该注解 public class LoginRequestMessageHandler extends SimpleChannelInboundHandler<LoginRequestMessage> { @Override protected void channelRead0(ChannelHandlerContext ctx, LoginRequestMessage msg) throws Exception { // 获得登录信息 String username = msg.getUsername(); String password = msg.getPassword(); // 校验登录信息 boolean login = UserServiceFactory.getUserService().login(username, password); LoginResponseMessage message; if (login) { message = new LoginResponseMessage(true, "登陆成功"); // 绑定channel与user SessionFactory.getSession().bind(ctx.channel(), username); } else { message = new LoginResponseMessage(false, "登陆失败"); } ctx.writeAndFlush(message); } }
添加到socketchannel中
// 该handler处理登录请求 LoginRequestMessageHandler loginRequestMessageHandler = new LoginRequestMessageHandler(); ch.pipeline().addLast(new LoginRequestMessageHandler());
登录之后，可以发送单条消息、创建群聊、群发消息、加入群聊、退出等等，可以自行查看代码。就是在服务端添加不同的handler，处理不同的消息。
空闲监测
1、连接假死
产生的原因
网络设备出现故障，例如网卡，机房等，底层的 TCP 连接已经断开了，但应用程序没有感知到，仍然占用着资源
公网网络不稳定，出现丢包。如果连续出现丢包，这时现象就是客户端数据发不出去，服务端也一直收不到数据，会白白地消耗资源
应用程序线程阻塞，无法进行数据读写
产生的问题
假死的连接占用的资源不能自动释放
向假死的连接发送数据，得到的反馈是发送超时
解决方法
可以添加IdleStateHandler对空闲时间进行检测，当指定时间内未发生读或写事件时，会触发特定事件。通过构造函数可以传入三个参数：
readerIdleTimeSeconds 读空闲经过的秒数
writerIdleTimeSeconds 写空闲经过的秒数
allIdleTimeSeconds 读和写空闲经过的秒数
服务端代码增加：
// 用于空闲连接的检测，5s内未读到数据，会触发READ_IDLE事件 ch.pipeline().addLast(new IdleStateHandler(5, 0, 0)); // 添加双向处理器，负责处理READER_IDLE事件 ch.pipeline().addLast(new ChannelDuplexHandler() { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { // 获得事件 IdleStateEvent event = (IdleStateEvent) evt; if (event.state() == IdleState.READER_IDLE) { // 断开连接 ctx.channel().close(); } } });
读空闲会触发READER_IDLE
写空闲会触发WRITE_IDLE
读和写空闲会触发ALL_IDEL
使用IdleStateHandler进行空闲检测，使用双向处理器ChannelDuplexHandler对入站与出站事件进行处理。
IdleStateHandler中的事件为特殊事件，需要实现ChannelDuplexHandler的userEventTriggered方法，判断事件类型并自定义处理方式，来对事件进行处理。
为避免因非网络等原因引发的READ_IDLE事件，比如网络情况良好，只是用户本身没有输入数据，这时发生READ_IDLE事件，直接让服务器断开连接是不可取的。
为避免此类情况，需要在客户端向服务器发送心跳包，发送频率要小于服务器设置的IdleTimeSeconds，一般设置为其值的一半。
客户端代码增加
// 发送心跳包，让服务器知道客户端在线 // 3s未发生WRITER_IDLE，就像服务器发送心跳包 // 该值为服务器端设置的READER_IDLE触发时间的一半左右 ch.pipeline().addLast(new IdleStateHandler(0, 3, 0)); ch.pipeline().addLast(new ChannelDuplexHandler() { @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { IdleStateEvent event = (IdleStateEvent) evt; if (event.state() == IdleState.WRITER_IDLE) { // 发送心跳包 ctx.writeAndFlush(new PingMessage()); } } });
详细代码已上传至码云：https://gitee.com/sglx666/netty-demo.git
参考链接：https://nyimac.gitee.io/2021/04/25/Netty%E5%9F%BA%E7%A1%80/