Netty在Android中使用

引用netty介绍文档中的一句话：Netty适合用来开发高性能长连接的客户端或服务器
其次netty基于NIO，并做了封装与优化以及一些高级算法，使得使用起来相比原生NIO更加容易,性能更好

timg.jpg

• 目录
• 1.使用方法
  • 1.1建立连接
  • 1.2数据发送
  • 1.3实现数据接收
• 2.常见问题处理及通信优化
  • 2.1心跳
  • 2.2重连接
  • 2.3数据接收不全
• 3.注意细节
• 4.其他

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);
                }

4.其他

• TCP粘包，拆包及解决方法
• netty处理分包粘包
• ByteBuf常用api