文章目录
- 三. Netty 进阶
- 1. 粘包与半包
- 1.1 ==粘包==现象
- 服务端代码
- 客户端代码
- 1.2 ==半包==现象
- 服务端代码
- 客户端代码
- 1.3 现象分析
- 粘包
- 半包
- 缘由
- 滑动窗口
- MSS 限制
- Nagle 算法
- 1.4 解决方案
- 方法1，短链接
- 方法2，固定长度
- 方法3，固定分隔符
- 方法4，预设长度
- 2. 协议设计与解析
- 2.1 为什么需要协议？
- 2.2 redis 协议举例
- 2.3 http 协议举例
- 2.4 自定义协议要素
- 编解码器
- MessageCodec
- TestMessageCodec
- 输出
- 💡 什么时候可以加 @Sharable
- 3. 聊天室案例
- 3.1 聊天室业务介绍
- 3.2 聊天室业务-登录
- 3.3 聊天室业务-单聊
- 3.4 聊天室业务-群聊
- 3.5 聊天室业务-退出
- 3.6 聊天室业务-空闲检测
- 连接假死
三. Netty 进阶
1. 粘包与半包
1.1 粘包现象
服务端代码
@Slf4j
public class HelloWorldServer {
void start() {
NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.group(boss, worker);
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.debug("connected {}", ctx.channel());
super.channelActive(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
log.debug("disconnect {}", ctx.channel());
super.channelInactive(ctx);
}
});
}
});
ChannelFuture channelFuture = serverBootstrap.bind(8080);
log.debug("{} binding...", channelFuture.channel());
channelFuture.sync();
log.debug("{} bound...", channelFuture.channel());
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("server error", e);
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
log.debug("stoped");
}
}
public static void main(String[] args) {
new HelloWorldServer().start();
}
}
客户端代码
客户端代码希望发送 10 个消息，每个消息是 16 字节
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class HelloWorldClient {
public static void main(String[] args) {
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
bootstrap.group(worker);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
log.debug("connected...");
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
// 会在连接 channel 建立成功之后，会触发 active事件
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.debug("sending...");
for (int i = 0; i < 10; i++) {
ByteBuf buffer = ctx.alloc().buffer();
buffer.writeBytes(new byte[]{0, 1, 2, 3, 4,
5, 6, 7, 8, 9,
10, 11, 12, 13, 14, 15});
ctx.writeAndFlush(buffer);
}
}
});
}
});
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("client error", e);
} finally {
worker.shutdownGracefully();
}
}
}
服务器端的某次输出，可以看到一次就接收了 160 个字节，而非分 10 次接收。而我们客户端是发了10次，每次16字节，这就是粘包现象。
// 客户端输出
/*
11:57:47 [DEBUG] [nioEventLoopGroup-2-1] c.z.a.HelloWorldClient - connetted...
11:57:47 [DEBUG] [nioEventLoopGroup-2-1] c.z.a.HelloWorldClient - sending...
*/
// 服务器端输出
/*
11:57:08 [DEBUG] [main] c.z.a.HelloWorldServer - [id: 0x6e5efdad] binding...
11:57:08 [DEBUG] [main] c.z.a.HelloWorldServer - [id: 0x6e5efdad, L:/0:0:0:0:0:0:0:0:8080] bound...
11:57:47 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xfaf43bd5, L:/127.0.0.1:8080 - R:/127.0.0.1:65248] REGISTERED
11:57:47 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xfaf43bd5, L:/127.0.0.1:8080 - R:/127.0.0.1:65248] ACTIVE
11:57:47 [DEBUG] [nioEventLoopGroup-3-1] c.z.a.HelloWorldServer - connected [id: 0xfaf43bd5, L:/127.0.0.1:8080 - R:/127.0.0.1:65248]
11:57:47 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xfaf43bd5, L:/127.0.0.1:8080 - R:/127.0.0.1:65248] READ: 160B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000010| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000020| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000030| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000040| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000050| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000060| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000070| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000080| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000090| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
+--------+-------------------------------------------------+----------------+
11:57:47 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0xfaf43bd5, L:/127.0.0.1:8080 - R:/127.0.0.1:65248] READ COMPLETE
*/
1.2 半包现象
服务端代码
只添加serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);这一行代码
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
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 HelloWorldServer {
void start() {
NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
// 为现象明显，服务端修改一下接收缓冲区，其它代码不变
// （调整系统的接收缓冲区-即滑动窗口）
serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.group(boss, worker);
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.debug("connected {}", ctx.channel());
super.channelActive(ctx);
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
log.debug("disconnect {}", ctx.channel());
super.channelInactive(ctx);
}
});
}
});
ChannelFuture channelFuture = serverBootstrap.bind(8080);
log.debug("{} binding...", channelFuture.channel());
channelFuture.sync();
log.debug("{} bound...", channelFuture.channel());
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("server error", e);
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
log.debug("stoped");
}
}
public static void main(String[] args) {
new HelloWorldServer().start();
}
}
客户端代码
客户端没有任何代码变动
@Slf4j
public class HelloWorldClient {
public static void main(String[] args) {
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
bootstrap.group(worker);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
log.debug("connected...");
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
// 会在连接 channel 建立成功之后，会触发 active事件
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.debug("sending...");
for (int i = 0; i < 10; i++) {
ByteBuf buffer = ctx.alloc().buffer();
buffer.writeBytes(new byte[]{0, 1, 2, 3, 4,
5, 6, 7, 8, 9,
10, 11, 12, 13, 14, 15});
ctx.writeAndFlush(buffer);
}
}
});
}
});
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("client error", e);
} finally {
worker.shutdownGracefully();
}
}
}
输出如下，可以看到客户端发了10次，每次16字节的消息，但是服务端分几次接受了36B、50B、50B、24B加起来共160字节。
（只要是使用TCP协议，那就会出现粘包和半包问题）
// 客户端输出
/*
12:10:42 [DEBUG] [nioEventLoopGroup-2-1] c.z.a.HelloWorldClient - connected...
12:10:42 [DEBUG] [nioEventLoopGroup-2-1] c.z.a.HelloWorldClient - sending...
*/
// 服务端输出
/*
12:10:32 [DEBUG] [main] c.z.a.HelloWorldServer - [id: 0xdafb9db3] binding...
12:10:32 [DEBUG] [main] c.z.a.HelloWorldServer - [id: 0xdafb9db3, L:/0:0:0:0:0:0:0:0:8080] bound...
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] REGISTERED
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] ACTIVE
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] c.z.a.HelloWorldServer - connected [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450]
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ: 36B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000010| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000020| 00 01 02 03 |.... |
+--------+-------------------------------------------------+----------------+
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ COMPLETE
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ: 50B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000010| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000020| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000030| 04 05 |.. |
+--------+-------------------------------------------------+----------------+
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ COMPLETE
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ: 50B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 04 05 |................|
|00000010| 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 04 05 |................|
|00000020| 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 04 05 |................|
|00000030| 06 07 |.. |
+--------+-------------------------------------------------+----------------+
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ COMPLETE
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ: 24B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 04 05 06 07 |................|
|00000010| 08 09 0a 0b 0c 0d 0e 0f |........ |
+--------+-------------------------------------------------+----------------+
12:10:42 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ COMPLETE
12:12:26 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] READ COMPLETE
12:12:26 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x07aed482, L:/127.0.0.1:8080 - R:/127.0.0.1:65450] EXCEPTION: java.io.IOException: 远程主机强迫关闭了一个现有的连接。
*/
注意
serverBootstrap.option(ChannelOption.SO_RCVBUF, 10) 影响的底层接收缓冲区（即滑动窗口）大小，仅决定了 netty 读取的最小单位，netty 实际每次读取的一般是它的整数倍
1.3 现象分析
粘包
-
现象，发送 abc def，接收 abcdef
-
原因
- 应用层（应用层面）：接收方 ByteBuf 设置太大（Netty 默认 1024）
- 滑动窗口（TCP层面）：假设发送方 256 bytes 表示一个完整报文，但由于接收方处理不及时且窗口大小足够大，这 256 bytes 字节就会缓冲在接收方的滑动窗口中，当滑动窗口中缓冲了多个报文就会粘包
- Nagle 算法（TCP层面）：会造成粘包（因为报文头也占据一定的大小，如果携带的实际数据较小，那显然不划算，因此会攒够了一批量数据，然后发送过去，因此也会造成粘包）
半包
- 现象，发送 abcdef，接收 abc def
- 原因
- 应用层（应用层面）：接收方 ByteBuf 小于实际发送数据量
- 滑动窗口（TCP层面）：假设接收方的窗口只剩了 128 bytes，发送方的报文大小是 256 bytes，这时放不下了，只能先发送前 128 bytes，等待 ack 后才能发送剩余部分，这就造成了半包
- MSS 限制（链路层面）：当发送的数据超过 MSS 限制后，会将数据切分发送，就会造成半包
本质是因为 TCP 是流式协议，消息无边界
缘由
滑动窗口
TCP 以一个段（segment）为单位，每发送一个段就需要进行一次确认应答（ack）处理，但如果这么做，缺点是包的往返时间越长性能就越差（因为tcp协议是可靠的传输协议，因此需要保证1次请求1次应答，但是如果等到上一次请求得到应答，再发送下一次请求，那效率显然就很低，因此下面引入了滑动窗口来解决这个问题）
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-leJ1X8om-1690552927974)(assets/0049.png)]
为了解决此问题，引入了窗口概念，窗口大小即决定了无需等待应答而可以继续发送的数据最大值
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-c4Pwo8wB-1690552927976)(assets/0051.png)]
窗口实际就起到一个缓冲区的作用，同时也能起到流量控制的作用
- 图中深色的部分即要发送的数据，高亮的部分即窗口
窗口内的数据才允许被发送，当应答未到达前，窗口必须停止滑动
如果 1001~2000 这个段的数据 ack 回来了，窗口就可以向前滑动
接收方也会维护一个窗口，只有落在窗口内的数据才能允许接收
MSS 限制
链路层对一次能够发送的最大数据有限制，这个限制称之为 MTU（maximum transmission unit），不同的链路设备的 MTU 值也有所不同，例如
以太网的 MTU 是 1500
FDDI（光纤分布式数据接口）的 MTU 是 4352
本地回环地址的 MTU 是 65535 - 本地测试不走网卡
MSS 是最大段长度（maximum segment size），它是 MTU 刨去 tcp 头和 ip 头后剩余能够作为数据传输的字节数
ipv4 tcp 头占用 20 bytes，ip 头占用 20 bytes，因此以太网 MSS 的值为 1500 - 40 = 1460
TCP 在传递大量数据时，会按照 MSS 大小将数据进行分割发送
MSS 的值在三次握手时通知对方自己 MSS 的值，然后在两者之间选择一个小值作为 MSS
Nagle 算法
- 即使发送一个字节，也需要加入 tcp 头和 ip 头，也就是总字节数会使用 41 bytes，非常不经济。因此为了提高网络利用率，tcp 希望尽可能发送足够大的数据，这就是 Nagle 算法产生的缘由
- 该算法是指发送端即使还有应该发送的数据，但如果这部分数据很少的话，则进行延迟发送
- 如果 SO_SNDBUF 的数据达到 MSS，则需要发送
- 如果 SO_SNDBUF 中含有 FIN（表示需要连接关闭）这时将剩余数据发送，再关闭
- 如果 TCP_NODELAY = true，则需要发送
- 已发送的数据都收到 ack 时，则需要发送
- 上述条件不满足，但发生超时（一般为 200ms）则需要发送
- 除上述情况，延迟发送
1.4 解决方案
- 短链接，发一个包建立一次连接，这样连接建立到连接断开之间就是消息的边界，缺点效率太低
- 每一条消息采用固定长度，缺点浪费空间
- 每一条消息采用分隔符，例如 \n，缺点需要转义
- 每一条消息分为 head 和 body，head 中包含 body 的长度
方法1，短链接
以解决粘包为例
public class HelloWorldClient {
static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
public static void main(String[] args) {
// 分 10 次发送
for (int i = 0; i < 10; i++) {
send();
}
}
private static void send() {
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
bootstrap.group(worker);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
log.debug("conneted...");
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
log.debug("sending...");
ByteBuf buffer = ctx.alloc().buffer();
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();
}
});
}
});
ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("client error", e);
} finally {
worker.shutdownGracefully();
}
}
}
输出，略
半包用这种办法还是解决不了，因为接收方的缓冲区大小是有限的（可再服务端通过如下方式调整netty的接收缓冲区，查看还是会出现半包现象：serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(16, 16, 16));
方法2，固定长度
客户端的操作就是一次发送一个100字节长度的消息, 其中每10个字节表示一条消息（不够10个字节的补位，凑够10个字节）
服务端使用定长解码器，每10个字节，凑够1条消息，才给到下一个handler
客户端代码
@Slf4j
public class HelloWorldClient {
public static void main(String[] args) {
send();
}
private static void send() {
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
bootstrap.group(worker);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
// 会在连接 channel 建立成功之后，会触发 active事件
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
/*
这里的操作就是发送一个100字节长度的消息, 其中每10个字节表示一条消息
*/
ByteBuf buf = ctx.alloc().buffer();
char c = '0';
Random r = new Random();
for (int i = 0; i < 10; i++) {
byte[] bytes = fill10Bytes(c, r.nextInt(10) + 1);
c++;
buf.writeBytes(bytes);
}
ctx.writeAndFlush(buf);
}
});
}
});
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("client error", e);
} finally {
worker.shutdownGracefully();
}
}
// 填够10个字节
public static byte[] fill10Bytes(char c, int len) {
byte[] bytes = new byte[10];
Arrays.fill(bytes, (byte) '_');
for (int i = 0; i < len; i++) {
bytes[i] = (byte) c;
}
System.out.println(new String(bytes));
return bytes;
}
}
服务端代码
@Slf4j
public class HelloWorldServer {
void start() {
NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
// 为现象明显，服务端修改一下接收缓冲区，其它代码不变
// (调整系统的接收缓冲区（即滑动窗口）)
// serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);
// 调整netty的接收缓冲区
// serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
// new AdaptiveRecvByteBufAllocator(16, 16, 16));
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.group(boss, worker);
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 注意顺序: 要先添加定长的解码处理器, 再添加日志处理器
ch.pipeline().addLast(new FixedLengthFrameDecoder(10));
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
}
});
ChannelFuture channelFuture = serverBootstrap.bind(8080);
log.debug("{} binding...", channelFuture.channel());
channelFuture.sync();
log.debug("{} bound...", channelFuture.channel());
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("server error", e);
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
log.debug("stoped");
}
}
public static void main(String[] args) {
new HelloWorldServer().start();
}
}
客户端输出，可以看到客户端输出，一次写入了100个字节
Connected to the target VM, address: '127.0.0.1:50935', transport: 'socket'
13:51:29 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xdc9dffac] REGISTERED
13:51:29 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xdc9dffac] CONNECT: /127.0.0.1:8080
13:51:29 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xdc9dffac, L:/127.0.0.1:50973 - R:/127.0.0.1:8080] ACTIVE
000000____
111_______
2222222222
333333____
444_______
55________
6666666___
77777_____
8888______
99999_____
13:51:29 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xdc9dffac, L:/127.0.0.1:50973 - R:/127.0.0.1:8080] WRITE: 100B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 30 30 30 30 30 30 5f 5f 5f 5f 31 31 31 5f 5f 5f |000000____111___|
|00000010| 5f 5f 5f 5f 32 32 32 32 32 32 32 32 32 32 33 33 |____222222222233|
|00000020| 33 33 33 33 5f 5f 5f 5f 34 34 34 5f 5f 5f 5f 5f |3333____444_____|
|00000030| 5f 5f 35 35 5f 5f 5f 5f 5f 5f 5f 5f 36 36 36 36 |__55________6666|
|00000040| 36 36 36 5f 5f 5f 37 37 37 37 37 5f 5f 5f 5f 5f |666___77777_____|
|00000050| 38 38 38 38 5f 5f 5f 5f 5f 5f 39 39 39 39 39 5f |8888______99999_|
|00000060| 5f 5f 5f 5f |____ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0xdc9dffac, L:/127.0.0.1:50973 - R:/127.0.0.1:8080] FLUSH
服务端输出，重点看服务端输出，服务端每10个字节接收作为一条消息，交给后面的handler处理，并没有发生半包
13:51:21 [DEBUG] [main] c.z.a.HelloWorldServer - [id: 0xafdaf078] binding...
13:51:21 [DEBUG] [main] c.z.a.HelloWorldServer - [id: 0xafdaf078, L:/0:0:0:0:0:0:0:0:8080] bound...
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] REGISTERED
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] ACTIVE
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 30 30 30 30 30 30 5f 5f 5f 5f |000000____ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 31 31 31 5f 5f 5f 5f 5f 5f 5f |111_______ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 32 32 32 32 32 32 32 32 32 32 |2222222222 |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 33 33 33 33 33 33 5f 5f 5f 5f |333333____ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 34 34 34 5f 5f 5f 5f 5f 5f 5f |444_______ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 35 35 5f 5f 5f 5f 5f 5f 5f 5f |55________ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 36 36 36 36 36 36 36 5f 5f 5f |6666666___ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 37 37 37 37 37 5f 5f 5f 5f 5f |77777_____ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 38 38 38 38 5f 5f 5f 5f 5f 5f |8888______ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ: 10B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 39 39 39 39 39 5f 5f 5f 5f 5f |99999_____ |
+--------+-------------------------------------------------+----------------+
13:51:29 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x0d23d3d4, L:/127.0.0.1:8080 - R:/127.0.0.1:50973] READ COMPLETE
缺点是，数据包的大小不好把握
- 长度定的太大，浪费
- 长度定的太小，对某些数据包又显得不够
方法3，固定分隔符
服务端代码
- 服务端加入，默认以 \n 或 \r\n 作为分隔符的处理器（如果超出最大长度仍未出现分隔符，则抛出异常）
@Slf4j
public class HelloWorldServer {
void start() {
NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
// 为现象明显，服务端修改一下接收缓冲区，其它代码不变
// (调整系统的接收缓冲区（即滑动窗口）)
// serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);
// 调整netty的接收缓冲区
// serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
// new AdaptiveRecvByteBufAllocator(16, 16, 16));
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.group(boss, worker);
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 注意顺序: 要先添加基于分隔符的解码处理器, 再添加日志处理器
ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
}
});
ChannelFuture channelFuture = serverBootstrap.bind(8080);
log.debug("{} binding...", channelFuture.channel());
channelFuture.sync();
log.debug("{} bound...", channelFuture.channel());
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("server error", e);
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
log.debug("stoped");
}
}
public static void main(String[] args) {
new HelloWorldServer().start();
}
}
客户端代码
客户端在每条消息之后，加入 \n 分隔符
@Slf4j
public class HelloWorldClient {
public static void main(String[] args) {
send();
}
private static void send() {
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
bootstrap.group(worker);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
// 会在连接 channel 建立成功之后，会触发 active事件
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
/*
在每条消息后面加上1个\n, 来表示消息的分隔符
*/
ByteBuf buf = ctx.alloc().buffer();
char c = '0';
Random r = new Random();
for (int i = 0; i < 10; i++) {
StringBuilder sb = makeString(c, r.nextInt(256) + 1);
buf.writeBytes(sb.toString().getBytes());
c++;
}
// 还是1次性写入buf
ctx.writeAndFlush(buf);
}
});
}
});
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("client error", e);
} finally {
worker.shutdownGracefully();
}
}
public static StringBuilder makeString(char c, int len) {
StringBuilder sb = new StringBuilder(len + 2);
for (int i = 0; i < len; i++) {
sb.append(c);
}
sb.append("\n");
return sb;
}
}
客户端输出
14:27:28 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x45f4846a] REGISTERED
14:27:28 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x45f4846a] CONNECT: /127.0.0.1:8080
14:27:28 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x45f4846a, L:/127.0.0.1:54341 - R:/127.0.0.1:8080] ACTIVE
14:27:28 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x45f4846a, L:/127.0.0.1:54341 - R:/127.0.0.1:8080] WRITE: 1445B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000010| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000020| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000030| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000040| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000050| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000060| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000070| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000080| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000090| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000a0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000b0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000c0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000d0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000e0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000f0| 30 30 30 30 0a 31 31 31 31 31 31 31 31 31 31 31 |0000.11111111111|
|00000100| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000110| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000120| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000130| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000140| 31 31 31 31 31 31 31 31 31 31 31 31 31 0a 32 32 |1111111111111.22|
|00000150| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000160| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000170| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000180| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000190| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000001a0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000001b0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000001c0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000001d0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000001e0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000001f0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000200| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000210| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000220| 32 32 0a 33 33 33 33 33 33 33 33 33 33 33 33 33 |22.3333333333333|
|00000230| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000240| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000250| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000260| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000270| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000280| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000290| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000002a0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000002b0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000002c0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000002d0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000002e0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000002f0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000300| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000310| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000320| 0a 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |.444444444444444|
|00000330| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000340| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000350| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000360| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000370| 34 34 0a 35 35 35 35 35 35 35 35 35 35 35 35 35 |44.5555555555555|
|00000380| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|00000390| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|000003a0| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|000003b0| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|000003c0| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|000003d0| 35 35 35 35 35 35 35 0a 36 36 36 36 36 36 36 36 |5555555.66666666|
|000003e0| 36 36 36 36 36 36 0a 37 37 37 37 37 37 37 37 37 |666666.777777777|
|000003f0| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000400| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000410| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000420| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000430| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000440| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000450| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000460| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000470| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000480| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000490| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000004a0| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000004b0| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000004c0| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000004d0| 37 0a 38 38 38 38 38 38 38 38 38 38 38 38 38 38 |7.88888888888888|
|000004e0| 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 |8888888888888888|
|000004f0| 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 |8888888888888888|
|00000500| 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 |8888888888888888|
|00000510| 38 38 38 38 38 38 38 0a 39 39 39 39 39 39 39 39 |8888888.99999999|
|00000520| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000530| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000540| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000550| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000560| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000570| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000580| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000590| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|000005a0| 39 39 39 39 0a |9999. |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x45f4846a, L:/127.0.0.1:54341 - R:/127.0.0.1:8080] FLUSH
服务端输出
14:27:25 [DEBUG] [main] c.z.a.HelloWorldServer - [id: 0xd4acc436] binding...
14:27:25 [DEBUG] [main] c.z.a.HelloWorldServer - [id: 0xd4acc436, L:/0:0:0:0:0:0:0:0:8080] bound...
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] REGISTERED
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] ACTIVE
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 244B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000010| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000020| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000030| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000040| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000050| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000060| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000070| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000080| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|00000090| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000a0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000b0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000c0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000d0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000e0| 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 30 |0000000000000000|
|000000f0| 30 30 30 30 |0000 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 88B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000010| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000020| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000030| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000040| 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 |1111111111111111|
|00000050| 31 31 31 31 31 31 31 31 |11111111 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 212B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000010| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000020| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000030| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000040| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000050| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000060| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000070| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000080| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|00000090| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000000a0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000000b0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000000c0| 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 32 |2222222222222222|
|000000d0| 32 32 32 32 |2222 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 253B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000010| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000020| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000030| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000040| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000050| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000060| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000070| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000080| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|00000090| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000000a0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000000b0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000000c0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000000d0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000000e0| 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333333|
|000000f0| 33 33 33 33 33 33 33 33 33 33 33 33 33 |3333333333333 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 81B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000010| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000020| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000030| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000040| 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 34 |4444444444444444|
|00000050| 34 |4 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 100B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|00000010| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|00000020| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|00000030| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|00000040| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|00000050| 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 35 |5555555555555555|
|00000060| 35 35 35 35 |5555 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 14B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 36 36 36 36 36 36 36 36 36 36 36 36 36 36 |66666666666666 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 234B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000010| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000020| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000030| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000040| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000050| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000060| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000070| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000080| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|00000090| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000000a0| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000000b0| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000000c0| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000000d0| 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 37 |7777777777777777|
|000000e0| 37 37 37 37 37 37 37 37 37 37 |7777777777 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 69B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 |8888888888888888|
|00000010| 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 |8888888888888888|
|00000020| 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 |8888888888888888|
|00000030| 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 38 |8888888888888888|
|00000040| 38 38 38 38 38 |88888 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ: 140B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000010| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000020| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000030| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000040| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000050| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000060| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000070| 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 39 |9999999999999999|
|00000080| 39 39 39 39 39 39 39 39 39 39 39 39 |999999999999 |
+--------+-------------------------------------------------+----------------+
14:27:28 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x8dfc81f2, L:/127.0.0.1:8080 - R:/127.0.0.1:54341] READ COMPLETE
缺点，处理字符数据比较合适，但如果内容本身包含了分隔符（字节数据常常会有此情况），那么就会解析错误
方法4，预设长度
在发送消息前，先约定用定长字节表示接下来数据的长度
LengthFieldBasedFrameDecoder
-
lengthFieldOffset - 长度字段偏移量
-
lengthFieldLength - 长度字段长度
-
lengthAdjustment - 长度字段为基准，还要跳过几个字节才是内容
-
initialBytesTostrip - 从头剥离几个字节
// 最大长度，长度偏移，长度占用字节，长度调整，剥离字节数
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 1, 0, 1));
测试示例，使用EmbeddedChannel模拟
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
public class TestLengthFieldDecoder {
public static void main(String[] args) {
// 注意处理器的添加顺序
EmbeddedChannel channel = new EmbeddedChannel(
new LengthFieldBasedFrameDecoder(1024, 0, 4, 1, 5),
new LoggingHandler(LogLevel.DEBUG)
);
// 4 个字节的内容长度， 实际内容
ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
// 模拟对方发送过来消息
send(buffer, "Hello, world");
send(buffer, "Hi!");
// 把消息写入channel
channel.writeInbound(buffer);
}
private static void send(ByteBuf buffer, String content) {
// 实际内容
byte[] bytes = content.getBytes();
// 实际内容长度
int length = bytes.length;
// 写入内容长度
buffer.writeInt(length);
// 写入版本号
buffer.writeByte(1);
// 写入实际内容
buffer.writeBytes(bytes);
}
}
// 输出
/*
16:09:07 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] REGISTERED
16:09:07 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] ACTIVE
16:09:07 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ: 12B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 2c 20 77 6f 72 6c 64 |Hello, world |
+--------+-------------------------------------------------+----------------+
16:09:07 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ: 3B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 69 21 |Hi! |
+--------+-------------------------------------------------+----------------+
16:09:07 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ COMPLETE
*/
2. 协议设计与解析
2.1 为什么需要协议？
TCP/IP 中消息传输基于流的方式，没有边界。
协议的目的就是划定消息的边界，制定通信双方要共同遵守的通信规则
例如：在网络上传输
下雨天留客天留我不留
是中文一句著名的无标点符号句子，在没有标点符号情况下，这句话有数种拆解方式，而意思却是完全不同，所以常被用作讲述标点符号的重要性
一种解读
下雨天留客，天留，我不留
另一种解读
下雨天，留客天，留我不？留
如何设计协议呢？其实就是给网络传输的信息加上“标点符号”。但通过分隔符来断句不是很好，因为分隔符本身如果用于传输，那么必须加以区分。因此，下面一种协议较为常用
定长字节表示内容长度 + 实际内容
例如，假设一个中文字符长度为 3，按照上述协议的规则，发送信息方式如下，就不会被接收方弄错意思了
0f下雨天留客06天留09我不留
小故事
很久很久以前，一位私塾先生到一家任教。双方签订了一纸协议：“无鸡鸭亦可无鱼肉亦可白菜豆腐不可少不得束修金”。此后，私塾先生虽然认真教课，但主人家则总是给私塾先生以白菜豆腐为菜，丝毫未见鸡鸭鱼肉的款待。私塾先生先是很不解，可是后来也就想通了：主人把鸡鸭鱼肉的钱都会换为束修金的，也罢。至此双方相安无事。
年关将至，一个学年段亦告结束。私塾先生临行时，也不见主人家为他交付束修金，遂与主家理论。然主家亦振振有词：“有协议为证——无鸡鸭亦可，无鱼肉亦可，白菜豆腐不可少，不得束修金。这白纸黑字明摆着的，你有什么要说的呢？”
私塾先生据理力争：“协议是这样的——无鸡，鸭亦可；无鱼，肉亦可；白菜豆腐不可，少不得束修金。”
双方唇枪舌战，你来我往，真个是不亦乐乎！
这里的束修金，也作“束脩”，应当是泛指教师应当得到的报酬
2.2 redis 协议举例
使用netty编写客户端，连接redis，并且发送set aaa bbb，get aaa 这2个命令
@Slf4j
public class TestRedis {
/*
set name zhangsan
*3 （3个元素 \r\n）
$3 （第一个元素3个字节 \r\n）
set （set 3个字节）
$4 （第二个元素4个字节 \r\n）
name（name 4个字节）
$8 （第三个元素8个字节 \r\n）
zhangsan（zhangsan 8个字节）
*/
public static void main(String[] args) {
NioEventLoopGroup worker = new NioEventLoopGroup();
byte[] LINE = new byte[]{13, 10};
System.out.println(LINE_BYTES[0] == '\r');
System.out.println(LINE_BYTES[1] == '\n');
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
bootstrap.group(worker);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new LoggingHandler());
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
// 会在连接 channel 建立成功后，会触发 active 事件
@Override
public void channelActive(ChannelHandlerContext ctx) {
// 执行命令: set aaa bbb
set(ctx);
// 执行命令: get aaa
get(ctx);
}
private void get(ChannelHandlerContext ctx) {
ByteBuf buf = ctx.alloc().buffer();
buf.writeBytes("*2".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("$3".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("get".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("$3".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("aaa".getBytes());
buf.writeBytes(LINE);
ctx.writeAndFlush(buf);
}
private void set(ChannelHandlerContext ctx) {
ByteBuf buf = ctx.alloc().buffer();
buf.writeBytes("*3".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("$3".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("set".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("$3".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("aaa".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("$3".getBytes());
buf.writeBytes(LINE);
buf.writeBytes("bbb".getBytes());
buf.writeBytes(LINE); // 注意：最后面的这个\r\n也不能省略
ctx.writeAndFlush(buf);
}
@Override
public void channelRead(ChannelHandlerContext ctx,
Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
System.out.println(buf.toString(Charset.defaultCharset()));
// 我觉得这里应该需要释放buf吧??
}
});
}
});
ChannelFuture channelFuture = bootstrap.connect("localhost", 6379).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
log.error("client error", e);
} finally {
worker.shutdownGracefully(); // 优雅关闭
}
}
}
// 输出日志
/*
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1] REGISTERED
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1] CONNECT: localhost/127.0.0.1:6379
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1, L:/127.0.0.1:58542 - R:localhost/127.0.0.1:6379] ACTIVE
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1, L:/127.0.0.1:58542 - R:localhost/127.0.0.1:6379] WRITE: 31B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 2a 33 0d 0a 24 33 0d 0a 73 65 74 0d 0a 24 33 0d |*3..$3..set..$3.|
|00000010| 0a 61 61 61 0d 0a 24 33 0d 0a 62 62 62 0d 0a |.aaa..$3..bbb.. |
+--------+-------------------------------------------------+----------------+
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1, L:/127.0.0.1:58542 - R:localhost/127.0.0.1:6379] FLUSH
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1, L:/127.0.0.1:58542 - R:localhost/127.0.0.1:6379] WRITE: 22B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 2a 32 0d 0a 24 33 0d 0a 67 65 74 0d 0a 24 33 0d |*2..$3..get..$3.|
|00000010| 0a 61 61 61 0d 0a |.aaa.. |
+--------+-------------------------------------------------+----------------+
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1, L:/127.0.0.1:58542 - R:localhost/127.0.0.1:6379] FLUSH
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1, L:/127.0.0.1:58542 - R:localhost/127.0.0.1:6379] READ: 14B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 2b 4f 4b 0d 0a 24 33 0d 0a 62 62 62 0d 0a |+OK..$3..bbb.. |
+--------+-------------------------------------------------+----------------+
+OK
$3
bbb
12:57:21 [DEBUG] [nioEventLoopGroup-2-1] i.n.h.l.LoggingHandler - [id: 0x80c5b3b1, L:/127.0.0.1:58542 - R:localhost/127.0.0.1:6379] READ COMPLETE
*/
2.3 http 协议举例
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class TestHttp {
public static void main(String[] args) throws InterruptedException {
NioEventLoopGroup boss = new NioEventLoopGroup(1);
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap()
.group(boss, worker)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
// 配置http编解码器（包括编码、解码）
// 解码 - 把发向服务端的请求ByteBuf解码
// 编码 - 把发给客户端的数据编码成ByteBuf, 以便写出字节数据
// （HttpServerCodec继承自CombinedChannelDuplexHandler<HttpRequestDecoder,
// HttpResponseEncoder>）
ch.pipeline().addLast(new HttpServerCodec());
// 注意写入的泛型, 表示该处理器只处理该类型的消息
ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
@Override
protected void channelRead0(ChannelHandlerContext ctx,
HttpRequest msg) throws Exception {
// 获取请求uri
log.info(msg.uri());
// 响应对象
DefaultFullHttpResponse response = new DefaultFullHttpResponse(
msg.protocolVersion(),
HttpResponseStatus.OK
);
byte[] content = "<h1>hello,world</h1>".getBytes();
response.content().writeBytes(content);
// 告诉浏览器, 响应内容就这么长, 不然浏览器会一直傻傻的等待接收后面的内容
// （网站图标位置一直在转圈圈）
response.headers().set(HttpHeaderNames.CONTENT_LENGTH,
content.length);
// 写回响应
// （这个response将会经过HttpServerCodec出站处理器，将它编码了，
// 再返回回去）
ctx.writeAndFlush(response);
}
});
/*
// 当浏览器发起1个get请求后, 该处理器的channelRead方法被调用了2次，
// - 第一次传过来的msg是 DefaultHttpRequest(decodeResult: success,
version: HTTP/1.1)
// - 第二次传过来的是 EmptyLastHttpContent
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelRead(ChannelHandlerContext ctx,
Object msg) throws Exception {
log.info("接收到请求: {}", msg);
// HttpServerCodec会把接收到的请求,
// 分为 HttpRequest 和 HttpContent(对该handler进行了2次调用)
// 如果只处理其中某一种类型的消息，可以如上使用
// SimpleChannelInboundHandler，
// 并指定泛型（比如上面就指定了HttpRequest）
if (msg instanceof HttpRequest) { // 请求行 和 请求头
log.info("HttpRequest msg: {}", msg);
} else if (msg instanceof HttpContent) { // 请求体
log.info("HttpContent msg: {}", msg);
}
}
});
*/
}
});
ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
}
}
}
2.4 自定义协议要素
- 魔数，用来在第一时间判定是否是无效数据包
- 版本号，可以支持协议的升级
- 序列化算法，消息正文到底采用哪种序列化反序列化方式，可以由此扩展，例如：json、protobuf、hessian、jdk
- 指令类型，是登录、注册、单聊、群聊… 跟业务相关
- 请求序号，为了双工通信，提供异步能力
- 正文长度
- 消息正文
编解码器
根据上面的要素，设计一个登录请求消息和登录响应消息，并使用 Netty 完成收发
MessageCodec
@Slf4j
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());
// 无意义，对齐填充，为了满2^n固定字节数, 正好凑够16个固定字节数
out.writeByte(0xff);
/* 将Message对象先转为字节数组(使用ByteArrayOutputStream + ObjectOutputStream),
然后把字节数组写到ByteBuf中 */
// 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);
log.info("message 编码完成...{}", msg);
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
// 读取 4 字节的魔数（读指针向后移动4位）
int magicNum = in.readInt();
// 读取 1 字节的版本
byte version = in.readByte();
// 读取 1 字节的序列化方式
byte serializerType = in.readByte();
// 读取 1 字节的指令类型
byte messageType = in.readByte();
// 读取 4 个字节 序列id
int sequenceId = in.readInt();
// 读取 无意义 的字节（将读指针向后移动1位）
in.readByte();
// 读取 长度
int length = in.readInt();
// 使用读取到的长度创建字节数组
byte[] bytes = new byte[length];
// 将ByteBuf读取到字节数组
in.readBytes(bytes, 0, length);
/* 这里将ByteBuf的数据, 通过jdk反序列化的方式, 转为java对象,
那就要保证这个数据一定是对象完整的序列化字节数据,
如果不是一个完整的数据（即出现半包的情况）, 那就废了。
因此，*/
// jdk反序列化 将字节数组 转为 Message对象
ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
Message message = (Message) ois.readObject();
log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType,
messageType, sequenceId, length);
log.debug("message 解码完成{}", message);
// 将消息添加到out集合当中
out.add(message);
}
}
TestMessageCodec
@Slf4j
public class TestMessageCodec {
public static void main(String[] args) throws Exception {
EmbeddedChannel channel = new EmbeddedChannel(
new LoggingHandler(),
// 参数的含义：数据的最大长度为1024个字节、从索引0开始,偏移12个字节开始,是实际内容长度信息、
// 实际内容长度占4个字节、不做调整(即不偏移,实际内容长度后面开始就是实际内容)、
// 不剥离前面的字节（因为在MessageCodec中有解析前面的字节）
// 帧解码器，解决粘包&半包问题
// （如果发现数据不完整(即发生了半包问题), 它是不会把消息传递给下一个处理器处理的,
// 等数据完整了，它才会把完整的消息发给下一个处理器）
// 此处可用来解决半包的问题: 也就是说, 实际内容长度可能有239个字节, 但是本次只收到了100个字节,
// 那如果往后面读的话, 就读不满239个字节, 那就等接收到下一次消息后,
// 一直到满了239个字节, 再把这个完整的消息, 传递给下一个handler
new LengthFieldBasedFrameDecoder(1024, 12, 4, 0, 0),
// 自定义编解码器
new MessageCodec()
);
log.info("------------------测试编码-----------------");
// 测试encode
LoginRequestMessage message = new LoginRequestMessage("zhangsan", "123");
// 向channel中写出数据（MessageCodec会对message作编码处理，写入到ByteBuf）
channel.writeOutbound(message);
log.info("------------------测试解码-----------------");
// 测试decode
// 先使用MessageCodec将message写入到ByteBuf
ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
new MessageCodec().encode(null, message, buf);
// 将buf切片成2部分, 模拟半包(先接收100字节, 然后再接收剩下的字节),
// 注意: 须使用LengthFieldBasedFrameDecoder处理半包问题
ByteBuf s1 = buf.slice(0, 100);
ByteBuf s2 = buf.slice(100, buf.readableBytes() - 100);
s1.retain(); // 引用计数+1（即现在为2）
// 这个方法默认会释放buf（release 1）,接收半包后的数据时,buf被释放了就会报错(下面这句会报错)
// 因此上面就使用了s1.retain()让引用计数+1了
channel.writeInbound(s1);
channel.writeInbound(s2);
}
}
输出
观察下面的输出
- 编码的时候, 写出了224个字节
- 解码的时候，接受了2次，第一次接受收了100个字节，第二次接收了124个字节，并且只出现了1次解码完成，这说明在出现了半包的情况下LengthFieldBasedFrameDecoder帮助我们解决了半包问题，并且它会接收一个完整的包后，才会传给后面的handler
- 可以留意一下，在写出的时候，第一行的字节数据 正好 符合 我们的编码规则哦
21:50:22 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] REGISTERED
21:50:22 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] ACTIVE
21:50:22 [INFO ] [main] c.i.a.c.TestMessageCodec - ------------------测试编码-----------------
21:50:22 [INFO ] [main] c.i.p.MessageCodec - message 编码完成...LoginRequestMessage(super=Message(sequenceId=0, messageType=0), username=zhangsan, password=123)
21:50:22 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] WRITE: 214B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 01 00 00 00 00 00 00 ff 00 00 00 c6 |................|
|00000010| ac ed 00 05 73 72 00 25 63 6e 2e 69 74 63 61 73 |....sr.%cn.itcas|
|00000020| 74 2e 6d 65 73 73 61 67 65 2e 4c 6f 67 69 6e 52 |t.message.LoginR|
|00000030| 65 71 75 65 73 74 4d 65 73 73 61 67 65 a0 3f 71 |equestMessage.?q|
|00000040| cb 31 45 b5 88 02 00 02 4c 00 08 70 61 73 73 77 |.1E.....L..passw|
|00000050| 6f 72 64 74 00 12 4c 6a 61 76 61 2f 6c 61 6e 67 |ordt..Ljava/lang|
|00000060| 2f 53 74 72 69 6e 67 3b 4c 00 08 75 73 65 72 6e |/String;L..usern|
|00000070| 61 6d 65 71 00 7e 00 01 78 72 00 19 63 6e 2e 69 |ameq.~..xr..cn.i|
|00000080| 74 63 61 73 74 2e 6d 65 73 73 61 67 65 2e 4d 65 |tcast.message.Me|
|00000090| 73 73 61 67 65 3d dd 19 a0 bc 07 47 cb 02 00 02 |ssage=.....G....|
|000000a0| 49 00 0b 6d 65 73 73 61 67 65 54 79 70 65 49 00 |I..messageTypeI.|
|000000b0| 0a 73 65 71 75 65 6e 63 65 49 64 78 70 00 00 00 |.sequenceIdxp...|
|000000c0| 00 00 00 00 00 74 00 03 31 32 33 74 00 08 7a 68 |.....t..123t..zh|
|000000d0| 61 6e 67 73 61 6e |angsan |
+--------+-------------------------------------------------+----------------+
21:50:22 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] FLUSH
21:50:22 [INFO ] [main] c.i.a.c.TestMessageCodec - ------------------测试解码-----------------
21:50:22 [INFO ] [main] c.i.p.MessageCodec - message 编码完成...LoginRequestMessage(super=Message(sequenceId=0, messageType=0), username=zhangsan, password=123)
21:50:22 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ: 100B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 01 02 03 04 01 00 00 00 00 00 00 ff 00 00 00 c6 |................|
|00000010| ac ed 00 05 73 72 00 25 63 6e 2e 69 74 63 61 73 |....sr.%cn.itcas|
|00000020| 74 2e 6d 65 73 73 61 67 65 2e 4c 6f 67 69 6e 52 |t.message.LoginR|
|00000030| 65 71 75 65 73 74 4d 65 73 73 61 67 65 a0 3f 71 |equestMessage.?q|
|00000040| cb 31 45 b5 88 02 00 02 4c 00 08 70 61 73 73 77 |.1E.....L..passw|
|00000050| 6f 72 64 74 00 12 4c 6a 61 76 61 2f 6c 61 6e 67 |ordt..Ljava/lang|
|00000060| 2f 53 74 72 |/Str |
+--------+-------------------------------------------------+----------------+
21:50:22 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ COMPLETE
21:50:22 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ: 114B
+-------------------------------------------------+
| 0 1 2 3 4 5 6 7 8 9 a b c d e f |
+--------+-------------------------------------------------+----------------+
|00000000| 69 6e 67 3b 4c 00 08 75 73 65 72 6e 61 6d 65 71 |ing;L..usernameq|
|00000010| 00 7e 00 01 78 72 00 19 63 6e 2e 69 74 63 61 73 |.~..xr..cn.itcas|
|00000020| 74 2e 6d 65 73 73 61 67 65 2e 4d 65 73 73 61 67 |t.message.Messag|
|00000030| 65 3d dd 19 a0 bc 07 47 cb 02 00 02 49 00 0b 6d |e=.....G....I..m|
|00000040| 65 73 73 61 67 65 54 79 70 65 49 00 0a 73 65 71 |essageTypeI..seq|
|00000050| 75 65 6e 63 65 49 64 78 70 00 00 00 00 00 00 00 |uenceIdxp.......|
|00000060| 00 74 00 03 31 32 33 74 00 08 7a 68 61 6e 67 73 |.t..123t..zhangs|
|00000070| 61 6e |an |
+--------+-------------------------------------------------+----------------+
21:50:22 [DEBUG] [main] c.i.p.MessageCodec - 16909060, 1, 0, 0, 0, 198
21:50:22 [DEBUG] [main] c.i.p.MessageCodec - message 解码完成LoginRequestMessage(super=Message(sequenceId=0, messageType=0), username=zhangsan, password=123)
21:50:22 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ COMPLETE
💡 什么时候可以加 @Sharable
- 当 handler 不保存状态时，就可以安全地在多线程下被共享
- 但要注意对于编解码器类，不能继承 ByteToMessageCodec 或 CombinedChannelDuplexHandler 父类，他们的构造方法对 @Sharable 有限制
- 如果能确保编解码器不会保存状态，可以继承 MessageToMessageCodec 父类
@Slf4j
@ChannelHandler.Sharable
/**
* 必须和 LengthFieldBasedFrameDecoder 一起使用，确保接到的 ByteBuf 消息是完整的
*/
public class MessageCodecSharable extends MessageToMessageCodec<ByteBuf, Message> {
@Override
protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {
ByteBuf out = ctx.alloc().buffer();
// 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);
outList.add(out);
}
@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);
out.add(message);
}
}
3. 聊天室案例
3.1 聊天室业务介绍
/**
* 用户管理接口
*/
public interface UserService {
/**
* 登录
* @param username 用户名
* @param password 密码
* @return 登录成功返回 true, 否则返回 false
*/
boolean login(String username, String password);
}
/**
* 会话管理接口
*/
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);
}
/**
* 聊天组会话管理接口
*/
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);
}
3.2 聊天室业务-登录
@Slf4j
public class ChatServer {
public static void main(String[] args) {
NioEventLoopGroup boss = new NioEventLoopGroup();
NioEventLoopGroup worker = new NioEventLoopGroup();
LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.group(boss, worker);
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ProcotolFrameDecoder());
ch.pipeline().addLast(LOGGING_HANDLER);
ch.pipeline().addLast(MESSAGE_CODEC);
ch.pipeline().addLast(new 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, "登录成功");
} else {
message = new LoginResponseMessage(false, "用户名或密码不正确");
}
ctx.writeAndFlush(message);
}
});
}
});
Channel channel = serverBootstrap.bind(8080).sync().channel();
channel.closeFuture().sync();
} catch (InterruptedException e) {
log.error("server error", e);
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
}
}
}
@Slf4j
public class ChatClient {
public static void main(String[] args) {
NioEventLoopGroup group = new NioEventLoopGroup();
LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();
CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);
AtomicBoolean LOGIN = new AtomicBoolean(false);
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class);
bootstrap.group(group);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ProcotolFrameDecoder());
// ch.pipeline().addLast(LOGGING_HANDLER);
ch.pipeline().addLast(MESSAGE_CODEC);
ch.pipeline().addLast("client handler", new ChannelInboundHandlerAdapter() {
// 接收响应消息
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
log.debug("msg: {}", msg);
if ((msg instanceof LoginResponseMessage)) {
LoginResponseMessage response = (LoginResponseMessage) msg;
if (response.isSuccess()) {
// 如果登录成功
LOGIN.set(true);
}
// 唤醒 system in 线程
WAIT_FOR_LOGIN.countDown();
}
}
// 在连接建立后触发 active 事件
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 负责接收用户在控制台的输入，负责向服务器发送各种消息
new Thread(() -> {
Scanner scanner = new Scanner(System.in);
System.out.println("请输入用户名:");
String username = scanner.nextLine();
System.out.println("请输入密码:");
String password = scanner.nextLine();
// 构造消息对象
LoginRequestMessage message = new LoginRequestMessage(username, password);
// 发送消息
ctx.writeAndFlush(message);
System.out.println("等待后续操作...");
try {
WAIT_FOR_LOGIN.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 如果登录失败
if (!LOGIN.get()) {
ctx.channel().close();
return;
}
while (true) {
System.out.println("==================================");
System.out.println("send [username] [content]");
System.out.println("gsend [group name] [content]");
System.out.println("gcreate [group name] [m1,m2,m3...]");
System.out.println("gmembers [group name]");
System.out.println("gjoin [group name]");
System.out.println("gquit [group name]");
System.out.println("quit");
System.out.println("==================================");
String command = scanner.nextLine();
String[] s = command.split(" ");
switch (s[0]){
case "send":
ctx.writeAndFlush(new ChatRequestMessage(username, s[1], s[2]));
break;
case "gsend":
ctx.writeAndFlush(new GroupChatRequestMessage(username, s[1], s[2]));
break;
case "gcreate":
Set<String> set = new HashSet<>(Arrays.asList(s[2].split(",")));
set.add(username); // 加入自己
ctx.writeAndFlush(new GroupCreateRequestMessage(s[1], set));
break;
case "gmembers":
ctx.writeAndFlush(new GroupMembersRequestMessage(s[1]));
break;
case "gjoin":
ctx.writeAndFlush(new GroupJoinRequestMessage(username, s[1]));
break;
case "gquit":
ctx.writeAndFlush(new GroupQuitRequestMessage(username, s[1]));
break;
case "quit":
ctx.channel().close();
return;
}
}
}, "system in").start();
}
});
}
});
Channel channel = bootstrap.connect("localhost", 8080).sync().channel();
channel.closeFuture().sync();
} catch (Exception e) {
log.error("client error", e);
} finally {
group.shutdownGracefully();
}
}
}
3.3 聊天室业务-单聊
服务器端将 handler 独立出来
登录 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) {
SessionFactory.getSession().bind(ctx.channel(), username);
message = new LoginResponseMessage(true, "登录成功");
} else {
message = new LoginResponseMessage(false, "用户名或密码不正确");
}
ctx.writeAndFlush(message);
}
}
单聊 handler
@ChannelHandler.Sharable
public class ChatRequestMessageHandler extends SimpleChannelInboundHandler<ChatRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ChatRequestMessage msg) throws Exception {
String to = msg.getTo();
Channel channel = SessionFactory.getSession().getChannel(to);
// 在线
if(channel != null) {
channel.writeAndFlush(new ChatResponseMessage(msg.getFrom(), msg.getContent()));
}
// 不在线
else {
ctx.writeAndFlush(new ChatResponseMessage(false, "对方用户不存在或者不在线"));
}
}
}
3.4 聊天室业务-群聊
创建群聊
@ChannelHandler.Sharable
public class GroupCreateRequestMessageHandler extends SimpleChannelInboundHandler<GroupCreateRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, GroupCreateRequestMessage msg) throws Exception {
String groupName = msg.getGroupName();
Set<String> members = msg.getMembers();
// 群管理器
GroupSession groupSession = GroupSessionFactory.getGroupSession();
Group group = groupSession.createGroup(groupName, members);
if (group == null) {
// 发生成功消息
ctx.writeAndFlush(new GroupCreateResponseMessage(true, groupName + "创建成功"));
// 发送拉群消息
List<Channel> channels = groupSession.getMembersChannel(groupName);
for (Channel channel : channels) {
channel.writeAndFlush(new GroupCreateResponseMessage(true, "您已被拉入" + groupName));
}
} else {
ctx.writeAndFlush(new GroupCreateResponseMessage(false, groupName + "已经存在"));
}
}
}
群聊
@ChannelHandler.Sharable
public class GroupChatRequestMessageHandler extends SimpleChannelInboundHandler<GroupChatRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, GroupChatRequestMessage msg) throws Exception {
List<Channel> channels = GroupSessionFactory.getGroupSession()
.getMembersChannel(msg.getGroupName());
for (Channel channel : channels) {
channel.writeAndFlush(new GroupChatResponseMessage(msg.getFrom(), msg.getContent()));
}
}
}
加入群聊
@ChannelHandler.Sharable
public class GroupJoinRequestMessageHandler extends SimpleChannelInboundHandler<GroupJoinRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, GroupJoinRequestMessage msg) throws Exception {
Group group = GroupSessionFactory.getGroupSession().joinMember(msg.getGroupName(), msg.getUsername());
if (group != null) {
ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群加入成功"));
} else {
ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群不存在"));
}
}
}
退出群聊
@ChannelHandler.Sharable
public class GroupQuitRequestMessageHandler extends SimpleChannelInboundHandler<GroupQuitRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, GroupQuitRequestMessage msg) throws Exception {
Group group = GroupSessionFactory.getGroupSession().removeMember(msg.getGroupName(), msg.getUsername());
if (group != null) {
ctx.writeAndFlush(new GroupJoinResponseMessage(true, "已退出群" + msg.getGroupName()));
} else {
ctx.writeAndFlush(new GroupJoinResponseMessage(true, msg.getGroupName() + "群不存在"));
}
}
}
查看成员
@ChannelHandler.Sharable
public class GroupMembersRequestMessageHandler extends SimpleChannelInboundHandler<GroupMembersRequestMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, GroupMembersRequestMessage msg) throws Exception {
Set<String> members = GroupSessionFactory.getGroupSession()
.getMembers(msg.getGroupName());
ctx.writeAndFlush(new GroupMembersResponseMessage(members));
}
}
3.5 聊天室业务-退出
@Slf4j
@ChannelHandler.Sharable
public class QuitHandler extends ChannelInboundHandlerAdapter {
// 当连接断开时触发 inactive 事件
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
SessionFactory.getSession().unbind(ctx.channel());
log.debug("{} 已经断开", ctx.channel());
}
// 当出现异常时触发
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
SessionFactory.getSession().unbind(ctx.channel());
log.debug("{} 已经异常断开 异常是{}", ctx.channel(), cause.getMessage());
}
}
3.6 聊天室业务-空闲检测
连接假死
原因
- 网络设备出现故障，例如网卡，机房等，底层的 TCP 连接已经断开了，但应用程序没有感知到，仍然占用着资源。
- 公网网络不稳定，出现丢包。如果连续出现丢包，这时现象就是客户端数据发不出去，服务端也一直收不到数据，就这么一直耗着
- 应用程序线程阻塞，无法进行数据读写
问题
- 假死的连接占用的资源不能自动释放
- 向假死的连接发送数据，得到的反馈是发送超时
服务器端解决
- 怎么判断客户端连接是否假死呢？如果能收到客户端数据，说明没有假死。因此策略就可以定为，每隔一段时间就检查这段时间内是否接收到客户端数据，没有就可以判定为连接假死
// 用来判断是不是 读空闲时间过长，或 写空闲时间过长
// 5s 内如果没有收到 channel 的数据，会触发一个 IdleState#READER_IDLE 事件
ch.pipeline().addLast(new IdleStateHandler(5, 0, 0));
// ChannelDuplexHandler 可以同时作为入站和出站处理器
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) {
log.debug("已经 5s 没有读到数据了");
ctx.channel().close();
}
}
});
客户端定时心跳
- 客户端可以定时向服务器端发送数据，只要这个时间间隔小于服务器定义的空闲检测的时间间隔，那么就能防止前面提到的误判，客户端可以定义如下心跳处理器
// 用来判断是不是 读空闲时间过长，或 写空闲时间过长
// 3s 内如果没有向服务器写数据，会触发一个 IdleState#WRITER_IDLE 事件
ch.pipeline().addLast(new IdleStateHandler(0, 3, 0));
// ChannelDuplexHandler 可以同时作为入站和出站处理器
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) {
// log.debug("3s 没有写数据了，发送一个心跳包");
ctx.writeAndFlush(new PingMessage());
}
}
});
