tcp为了提高效率会将多个数据包批量发送过来,所以就会存在粘包和拆包的问题。使用Netty,可以通过自定义协议+编解码器来解决粘包和拆包的问题。
1.定义协议
public class MessageProtocol {
private int len;
private byte[] content;
public int getLen() {
return len;
}
public void setLen(int len) {
this.len = len;
}
public byte[] getContent() {
return content;
}
public void setContent(byte[] content) {
this.content = content;
}
}
2.自定义编码器和解码器
public class MyMessageEncoder extends MessageToByteEncoder<MessageProtocol> {
@Override
protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {
System.out.println("编码器被调用");
out.writeInt(msg.getLen());
out.writeBytes(msg.getContent());
}
}
public class MyMessageDecoder extends ReplayingDecoder<Void> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
int len = in.readInt();
byte[] bytes = new byte[len];
in.readBytes(bytes);
MessageProtocol messageProtocol = new MessageProtocol();
messageProtocol.setLen(len);
messageProtocol.setContent(bytes);
out.add(messageProtocol);
}
}
3.客户端程序
public class NettyClient {
public static void main(String[] args) throws Exception{
NioEventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMessageEncoder());
pipeline.addLast(new MyMessageDecoder());
pipeline.addLast(new NettyClientHandler());
}
});
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7000).sync();
channelFuture.channel().closeFuture().sync();
}finally {
group.shutdownGracefully();
}
}
}
public class NettyClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for (int i = 0; i < 10; i++){
String message = "今天天气冷,吃火锅";
byte[] bytes = message.getBytes("utf-8");
int len = bytes.length;
MessageProtocol messageProtocol = new MessageProtocol();
messageProtocol.setLen(len);
messageProtocol.setContent(bytes);
ctx.writeAndFlush(messageProtocol);
}
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
4.编写服务端
public class NettyServer {
public static void main(String[] args) throws Exception{
NioEventLoopGroup parentGroup = new NioEventLoopGroup(1);
NioEventLoopGroup childGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(parentGroup, childGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler())
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMessageDecoder());
pipeline.addLast(new MyMessageEncoder());
pipeline.addLast(new NettyServerHandler());
}
});
ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
channelFuture.channel().closeFuture().sync();
}finally {
parentGroup.shutdownGracefully();
childGroup.shutdownGracefully();
}
}
}
public class NettyServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
private int count;
@Override
protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
String message = new String(msg.getContent(), CharsetUtil.UTF_8);
System.out.println("服务端接收到数据" + message);
System.out.println("服务端接收数据长度" + msg.getLen());
System.out.println("接收次数" + (++this.count));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}