引用netty介绍文档中的一句话：Netty适合用来开发高性能长连接的客户端或服务器
其次netty基于NIO，并做了封装与优化以及一些高级算法，使得使用起来相比原生NIO更加容易,性能更好。
timg.jpg
1.使用方法
- 1.1建立连接
//进行初始化
NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(); //初始化线程组
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioSocketChannel.class).group(nioEventLoopGroup);
bootstrap.option(ChannelOption.TCP_NODELAY, true); //无阻塞
bootstrap.option(ChannelOption.SO_KEEPALIVE, true); //长连接
bootstrap.option(ChannelOption.SO_TIMEOUT, SLEEP_TIME); //收发超时
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(new ByteArrayDecoder()) //接收解码方式
.addLast(new ByteArrayEncoder()) //发送编码方式
.addLast(new ChannelHandle(ConnectService.this)); //处理数据接收
}
});
//开始建立连接并监听返回
ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress(AppInfo.serverIp, AppInfo.serverPort));
channelFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
AppInfo.isConnected = future.isSuccess();
if (future.isSuccess()) {
Log.d(TAG, "connect success !");
} else {
Log.d(TAG, "connect failed !");
}
}
});
- 1.2数据发送
private void sendDataToServer(byte[] sendBytes) {
if (sendBytes != null && sendBytes.length > 0) {
if (channelFuture != null && channelFuture.channel().isActive()) {
channelFuture.channel().writeAndFlush(sendBytes);
}
}
}
- 1.3实现数据接收类：
public static class ChannelHandle extends SimpleChannelInboundHandler<SimpleProtocol> {
private ConnectService connectService;
public ChannelHandle(ConnectService connectService) {
this.connectService = connectService;
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
Log.e(TAG, "channelInactive 连接失败");
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
super.userEventTriggered(ctx, evt);
if (evt instanceof IdleStateEvent) {
IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
if (idleStateEvent.state().equals(IdleState.WRITER_IDLE)) {
Log.d(TAG, "userEventTriggered write idle");
if (connectService == null){
return;
}
}else if (idleStateEvent.state().equals(IdleState.READER_IDLE)){
Log.d(TAG, "userEventTriggered read idle");
ctx.channel().close();
}
}
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, byte[] data) throws Exception {
if (simpleProtocol == null){
return;
}
//处理接收到的数据data
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
Log.i(TAG, "exceptionCaught");
cause.printStackTrace();
ctx.close();
}
}
2.常见问题处理及通信优化
-
2.1心跳
在初始化时添加：.addLast(new IdleStateHandler(30, 10, 0)) //参数1：代表读套接字超时的时间，例如30秒没收到数据会触发读超时回调;参数2：代表写套接字超时时间，例如10秒没有进行写会触发写超时回调;参数3：将在未执行读取或写入时触发超时回调，0代表不处理；读超时尽量设置大于写超时代表多次写超时时写心跳包，多次写了心跳数据仍然读超时代表当前连接错误，即可断开连接重新连接
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(new ByteArrayDecoder()) //接收解码方式
.addLast(new ByteArrayEncoder()) //发送编码方式
.addLast(new ChannelHandle(ConnectService.this)); //处理数据接收
.addLast(new IdleStateHandler(30, 10, 0))
}
});
在添加初始化.addLast(new ChannelHandle(ConnectService.this));的ChannelHandle 类中重写超时回调方法
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
super.userEventTriggered(ctx, evt);
if (evt instanceof IdleStateEvent) {
IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
if (idleStateEvent.state().equals(IdleState.WRITER_IDLE)) {
//写超时，此时可以发送心跳数据给服务器
Log.d(TAG, "userEventTriggered write idle");
if (connectService == null){
return;
}
}else if (idleStateEvent.state().equals(IdleState.READER_IDLE)){
//读超时，此时代表没有收到心跳返回可以关闭当前连接进行重连
Log.d(TAG, "userEventTriggered read idle");
ctx.channel().close();
}
}
}
- 2.2重连接
private void reConnect(){
if (channelFuture != null && channelFuture.channel() != null && channelFuture.channel().isActive()){
channelFuture.channel().close();//已经连接时先关闭当前连接，关闭时回调exceptionCaught进行重新连接
}else {
connect(); //当前未连接，直接连接即可
}
}
连接失败回调
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
super.channelInactive(ctx);
Log.e(TAG, "channelInactive 连接失败");
if (connectService != null) {
connectService.connect(); //重新连接
}
}
连接错误回调
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
Log.i(TAG, "exceptionCaught");
cause.printStackTrace();
ctx.close(); //关闭连接后回调channelInactive会重新调用connectService.connect();
}
-
2.3数据接收不全
出现问题的原因1：接收数据的缓冲区太小 解决办法：
bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(5000, 5000, 8000)); //接收缓冲区 最小值太小时数据接收不全
出现问题的原因2：没有处理粘包拆包问题 解决方法：
添加包的接收处理：addLast(new SimpleProtocolDecoder()) 解包处理避免粘包的主要步骤是：1.解协议头 2.解包长 3.读指定的包长后返回
public class SimpleProtocolDecoder extends ByteToMessageDecoder{
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
if (buffer.readableBytes() >= SimpleProtocol.BASE_LENGTH){
int beginReader;
while (true){
beginReader = buffer.readerIndex();
buffer.markReaderIndex();
if (buffer.readShort() == AppInfo.HEAD_TAG){
buffer.readByte();
break;
}
buffer.resetReaderIndex();
buffer.readByte();
if (buffer.readableBytes() < SimpleProtocol.BASE_LENGTH){
return;
}
}
int length = MyUtil.shortToBig(buffer.readShort()) - SimpleProtocol.BASE_LENGTH;
//buffer.readerIndex(beginReader);
//Log.e("TAG","length:"+buffer.readableBytes());
if (buffer.readableBytes() < length){
buffer.readerIndex(beginReader);
return;
}
//byte[] data = new byte[length];
byte[] data = new byte[length + SimpleProtocol.BASE_LENGTH];
int pos = buffer.readerIndex() - SimpleProtocol.BASE_LENGTH;
if (pos < 0){
return;
}
buffer.readerIndex(pos);
buffer.readBytes(data);
SimpleProtocol simpleProtocol = new SimpleProtocol((short) data.length,data);
out.add(simpleProtocol);
}
}
}
3.注意细节
- 把连接，读写都放到service中，模块封装方便各个组件调用
- 每次连接时使用线程池，避免ANR，节省开销
- 重新连接时不要对netty重复初始化，同时可以使用handler控制重连间隔避免短时间内死循环重复调用
- 重连时close当前channel的同时需要调用以下代码：
if (channelFuture != null && channelFutureListener != null){
channelFuture.removeListener(channelFutureListener);
channelFuture.cancel(true);
}
