Netty拆包粘包问题解决——特殊结束符

客户端和服务器&＃xff0c;协商定义一个特殊的分隔符号&＃xff0c;分隔符号长度自定义。如&＃xff1a;‘#’、‘$_$’、‘AA&＃64;’。在通讯的时候&＃xff0c;只要没有发送分隔符号&＃xff0c;则代表一条数据没有结束。

服务端

server

package com.server;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
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.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;import java.nio.charset.Charset;//服务器端
public class MyServer {//监听线程组&＃xff0c;监听客户端请求private EventLoopGroup acceptorGroup &＃61; null;//处理客户端相关操作线程组&＃xff0c;负责处理与客户端的数据通信private EventLoopGroup clientGroup &＃61; null;//服务启动相关配置信息,服务端Bootstrap带serverprivate ServerBootstrap serverBootstrap &＃61; null;public MyServer(){init();}private void init(){//初始化线程组,构建线程组的时候&＃xff0c;如果不传递参数&＃xff0c;则默认构建的线程组线程数是CPU核心数量acceptorGroup &＃61; new NioEventLoopGroup();clientGroup &＃61; new NioEventLoopGroup();//初始化服务的配置serverBootstrap &＃61; new ServerBootstrap();//绑定线程组&＃xff0c;acceptorGroup监听信息&＃xff0c;clientGroup客户端信息serverBootstrap.group(acceptorGroup,clientGroup);//设定通信模式为NIO&＃xff0c;同步非阻塞serverBootstrap.channel(NioServerSocketChannel.class);//设定缓冲区大小&＃xff0c;缓冲区单位是字节serverBootstrap.option(ChannelOption.SO_BACKLOG,1024);//SO_SNDBUF发送缓冲区&＃xff0c;SO_RCVBUF接收缓冲区&＃xff0c;SO_KEEPALIVE开启心跳监测&＃xff08;保证连接有效&＃xff09;serverBootstrap.option(ChannelOption.SO_SNDBUF,16*1024).option(ChannelOption.SO_RCVBUF,16*1024).option(ChannelOption.SO_KEEPALIVE,true);}public ChannelFuture doAccept(int port) throws InterruptedException {/*** childHandler是服务的bootstrap独有的方法&＃xff0c;用于提供处理对象* 可以一次性增加若干个处理逻辑&＃xff0c;是类似责任链模式的处理方式* 增加A,B两个处理逻辑&＃xff0c;在处理客户端请求数据的时候&＃xff0c;根据A->B顺序依次处理*/serverBootstrap.childHandler(new ChannelInitializer() {&＃64;Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {//数据分隔符&＃xff0c;定义的数据分隔符一定是一个ByteBuf类型的数据对象ByteBuf delimiter &＃61; Unpooled.copiedBuffer("$E$".getBytes());ChannelHandler[] acceptorHandlers &＃61; new ChannelHandler[3];//处理固定结束标记符号的Handler&＃xff0c;这个Handler没有&＃64;Sharabler注解修饰//必须每次初始化通道时创建一个新的对象//使用特殊符号分隔处理数据粘包问题&＃xff0c;也要定义每个数据包最大长度&＃xff0c;Netty建议数据有最大长度acceptorHandlers[0] &＃61; new DelimiterBasedFrameDecoder(1024,delimiter);//字符串解码器Handler&＃xff0c;会自动处理channelRead方法的msg参数&＃xff0c;将ByteBuf类型的数据转换成字符串acceptorHandlers[1] &＃61; new StringDecoder(Charset.forName("utf-8"));acceptorHandlers[2] &＃61; new MyServerHandler();socketChannel.pipeline().addLast(acceptorHandlers);}});/*** bind方法 - 绑定监听端口&＃xff0c;serverBootstrap可以绑定多个监听端口&＃xff0c;多次调用即可* sync - 开始监听逻辑&＃xff0c;返回一个ChannelFuture&＃xff0c;返回结果代表的是监听成功后的一个对应的未来结果* 可以使用ChannelFuture实现后续的服务器和客户端的交互*/ChannelFuture future &＃61; serverBootstrap.bind(port).sync();/*绑定多个端口serverBootstrap.bind(port);serverBootstrap.bind(port);*/return future;}/*** shutdownGracefully - 是一个安全关闭的方法&＃xff0c;可以保证不放弃任何一个已接收的客户端请求*/public void release(){this.acceptorGroup.shutdownGracefully();this.clientGroup.shutdownGracefully();}public static void main(String[] args){ChannelFuture future &＃61; null;MyServer myServer &＃61; null;try {myServer &＃61; new MyServer();future &＃61; myServer.doAccept(9999);System.out.println("server started");//关闭连接future.channel().closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();}finally {if (null !&＃61; future){try {future.channel().closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();}}if (null !&＃61; myServer){myServer.release();}}}
}


serverHandler

package com.server;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;import java.io.UnsupportedEncodingException;/*** &＃64;Sharable代表当前Handler是一个可以分享的处理器&＃xff0c;可以分享给多个客户端同时使用* 如不使用注解类型&＃xff0c;每次客户请求时&＃xff0c;必须为客户重新创建一个新的Handler对象*/
&＃64;Sharable
public class MyServerHandler extends ChannelHandlerAdapter{/*** 业务处理逻辑* 用于处理读取数据请求的逻辑* ctx - 上下文对象&＃xff0c;其中包含于客户端建立连接的所有资源&＃xff0c;如&＃xff1a;对应的Channel* msg - 读取到的数据&＃xff0c;默认类型是ByteBuf,是Netty自定义的&＃xff0c;是对ByteBuffer的封装&＃xff0c;不用考虑复位问题*/&＃64;Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws UnsupportedEncodingException {String message &＃61; msg.toString();System.out.println("from client :"&＃43;message);String line &＃61; "server message $E$ test delimiter handler!! $E$ second message $E$";if ("exit".equals(message)){ctx.close();return;}//写操作自动释放缓存&＃xff0c;避免内存溢出ctx.writeAndFlush(Unpooled.copiedBuffer(line.getBytes("utf-8")));/*如果调用的是write方法&＃xff0c;不会刷新缓存&＃xff0c;缓存中的数据不会发送到客户端&＃xff0c;必须再次调用flush方法才行ctx.write(Unpooled.copiedBuffer(line.getBytes("utf-8")));ctx.close();*/}/*** 异常处理逻辑,当客户端异常退出时也会执行* ChannelHandlerContext关闭&＃xff0c;也代表当前与客户端连接资源关闭*/&＃64;Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){System.out.println("server exceptionCaught method run..");ctx.close();}
}


客户端

client

package com.client;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;import java.nio.charset.Charset;
import java.util.Scanner;
import java.util.Timer;
import java.util.concurrent.TimeUnit;/*** 客户端是请求的发起者&＃xff0c;不需要监听* 只需要定义唯一的一个线程组即可*/
public class CustorClient {//处理请求和处理服务端响应的线程组private EventLoopGroup group &＃61; null;//客户端服务启动相关配置信息private Bootstrap bootstrap &＃61; null;public CustorClient(){init();}private void init(){group &＃61; new NioEventLoopGroup();bootstrap &＃61; new Bootstrap();//绑定线程组bootstrap.group(group);//设定通讯模式为NIObootstrap.channel(NioSocketChannel.class);}public ChannelFuture doRequest(String host, int port) throws InterruptedException {/*** 客户端的bootstrap没有childHandler方法&＃xff0c;只有handler方法* 方法含义等同于ServerBootstrap中的childHandler* 在客户端必须绑定处理器&＃xff08;必须调用handler方法&＃xff09;* 服务器必须绑定处理器&＃xff08;必须调用childHandler方法&＃xff09;*/this.bootstrap.handler(new ChannelInitializer() {&＃64;Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {//数据分隔符ByteBuf delimiter &＃61; Unpooled.copiedBuffer("$E$".getBytes());ChannelHandler[] handlers &＃61; new ChannelHandler[3];handlers[0] &＃61; new DelimiterBasedFrameDecoder(1024,delimiter);//字符串解码器handlerhandlers[1] &＃61; new StringDecoder(Charset.forName("utf-8"));handlers[2] &＃61; new CustorClientHandler();socketChannel.pipeline().addLast(handlers);}});//建立连接ChannelFuture future &＃61; this.bootstrap.connect(host,port).sync();return future;}public void release(){this.group.shutdownGracefully();}public static void main(String[] atgs){CustorClient client &＃61; null;ChannelFuture future &＃61; null;try {client &＃61; new CustorClient();future &＃61; client.doRequest("localhost", 9999);Scanner s &＃61; null;while (true) {s &＃61; new Scanner(System.in);System.out.println("enter message send to server(enter &＃39;exit&＃39; for close client)");String line &＃61; s.nextLine();if ("exit".equals(line)) {/*** addListener - 增加监听&＃xff0c;当条件满足时候&＃xff0c;出发监听器* ChannelFutureListener.CLOSE - 关闭监听器&＃xff0c;代表ChannelFuture执行返回后&＃xff0c;关闭连接*/future.channel().writeAndFlush(Unpooled.copiedBuffer(line.getBytes("utf-8"))).addListener(ChannelFutureListener.CLOSE);break;}//Unpooled工具类用来做buffer转换future.channel().writeAndFlush(Unpooled.copiedBuffer(line.getBytes("utf-8")));//睡一秒读取信息TimeUnit.SECONDS.sleep(1);}}catch (Exception e){e.printStackTrace();}finally {if (null !&＃61; future){try {future.channel().closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();}}if (null !&＃61; client){client.release();}}}
}


clientHandler

package com.client;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;import java.io.UnsupportedEncodingException;public class CustorClientHandler extends ChannelHandlerAdapter{&＃64;Overridepublic void channelRead(ChannelHandlerContext cxt, Object msg) throws UnsupportedEncodingException {try {String message &＃61; msg.toString();System.out.println("form server:"&＃43;message);} finally {//释放资源&＃xff0c;避免内存溢出ReferenceCountUtil.release(msg);}}&＃64;Overridepublic void exceptionCaught(ChannelHandlerContext cxt,Throwable cause){System.out.println("client exceptionCaught method run..");cxt.close();}
}


测试