Netty的重连机制实现

前言

Netty心跳重连的代码，需要解决以下几个问题

1）ChannelPipeline中的ChannelHandlers的维护，首次连接和重连都需要对ChannelHandlers进行管理
2）重连对象的管理，也就是bootstrap对象的管理
3）重连机制编写

下面是代码实现...

一.客户端

重连机制主要是在客户端来实现，下面直接上代码

public class HeartClient {
    public static void main(String[] args){
        new HeartClient().init("localhost",8080);
    }

    public void init(String address,int port){
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(1);
        Bootstrap bootstrap = new Bootstrap();

        //触发发送心跳包的类
        ConnectorIdleStateTrigger idleStateTrigger = new ConnectorIdleStateTrigger();

        //netty提供的HashedWheelTimer 主要用来高效处理大量定时任务
        // 且任务对时间精度要求相对不高, 比如链接超时管理等场景, 缺点是内存占用相对较高.
        HashedWheelTimer timer = new HashedWheelTimer();
        bootstrap.group(workerGroup).
                option(ChannelOption.TCP_NODELAY,true).
                channel(NioSocketChannel.class).
                handler(new LoggingHandler());

        //顾名思义 监视连接的类，重连机制主要靠他
        ConnectionWatchdog connectionWatchdog = new ConnectionWatchdog(bootstrap,timer,address,port) {
            @Override
            public ChannelHandler[] handler() {
                return new ChannelHandler[]{
                        this,
                        new IdleStateHandler(0, 5, 0),
                        new MessageEncoder(),
                        new MessageDecoder(1 << 20, 9, 4, 0, 0, false),
                        idleStateTrigger,
                        new HeartClientHanlder()
                };
            }
        };
        ChannelFuture future;
        try {
            synchronized (bootstrap){
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(connectionWatchdog.handler());
                    }
                });
                //进行连接
                future = bootstrap.connect(new InetSocketAddress(address,port));
            }
            future.sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

主要用于重连的时候复用ChannelHandler

public interface ChannelHandlerHolder {
    ChannelHandler[] handler();
}

最最关键的来了，ConnectionWatchdog 可以去观察链路是否断了，如果断了，进行循环的断线重连操作

@ChannelHandler.Sharable
public abstract class ConnectionWatchdog extends ChannelInboundHandlerAdapter implements ChannelHandlerHolder , TimerTask {
    //尝试次数
    private  int attempts;
    //bootstrap对象，重连的时候依旧需要这个对象
    private Bootstrap bootstrap;
    //是否重连
    boolean reconnect = true;
    //执行重连任务的调度器
    private Timer timer;
    //地址 端口号
    private String address;
    private int port;

    public ConnectionWatchdog(Bootstrap bootstrap, Timer timer, String address, int port) {
        this.bootstrap = bootstrap;
        this.timer = timer;
        this.address = address;
        this.port = port;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("当前channel以及激活,尝试次数重置为0");
        attempts=0;
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("当前channel关闭");
        if (reconnect){

            if (attempts<12){
                attempts++;
                int timeouts = 1<<attempts;
                System.out.println("正在尝试重新建立连接:"+"第"+attempts+"次");
                timer.newTimeout(this,timeouts, TimeUnit.SECONDS);
            }
        }
        ctx.fireChannelInactive();
    }


    @Override
    public void run(Timeout timeout) throws Exception {
        ChannelFuture future = null;
        //线程同步 
        synchronized (bootstrap){
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast(handler());
                }
            });
            future = bootstrap.connect(new InetSocketAddress(this.address, this.port));
        }
        //增加监听器 判断是否成功
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                //判断是否重新连接成功
                boolean successd = channelFuture.isSuccess();
                if (successd){
                    System.out.println("重新连接成功");
                }else{
                    //如果重连失败，则调用ChannelInactive方法，再次出发重连事件，一直尝试12次，如果失败则不再重连
                    channelFuture.channel().pipeline().fireChannelInactive();
                }
            }
        });
    }
}

定时发送心跳包

@ChannelHandler.Sharable
public class ConnectorIdleStateTrigger extends ChannelInboundHandlerAdapter {

    private final Message heartMsg = new Message((byte)Integer.parseInt("AF",16), System.currentTimeMillis(),"Heartbeat");

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent){
            IdleStateEvent stateEvent = (IdleStateEvent) evt;
            IdleState state = stateEvent.state();
            if (state == IdleState.WRITER_IDLE){
                ctx.writeAndFlush(heartMsg);
            }
        }else{
            super.userEventTriggered(ctx, evt);
        }
    }
}

这里是客户端用于处理和服务端的通信

@ChannelHandler.Sharable
public class HeartClientHanlder extends ChannelInboundHandlerAdapter {
    private Scanner scanner = new Scanner(System.in);;
    private ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
    private ExecutorService executorService = Executors.newFixedThreadPool(10);;
    private Thread thread;
    private FutureTask futureTask;
    //表示业务包 非心跳包
    private final int flag = Integer.parseInt("CF", 16);
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client active:"+new Date());
        ctx.writeAndFlush(new Message((byte)flag,System.currentTimeMillis(),"hello,I am common client!"));
        ctx.fireChannelActive();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Message message = (Message) msg;
        System.out.println("from server >>>"+message.getType()+":"+message.getRequestId()+":"+message.getBody());
        //这里异步执行，才能看到效果，不然线程都被scanner阻塞住了
        singleThreadExecutor.execute(()->{
            //这里再套一个线程，是用来处理scanner从控制台读取信息的
            this.futureTask = new FutureTask<>(new MsgThread(ctx));
            this.executorService.execute(futureTask);
            try {
                //这个get方法会阻塞线程，所有才在外面又套了一层，完成异步操作
                this.futureTask.get();
            } catch (Exception e) {
            }
        });

    }

    class MsgThread implements Callable {
        ChannelHandlerContext ctx;

        public MsgThread(ChannelHandlerContext ctx) {
            this.ctx = ctx;
        }

        @Override
        public Object call() throws Exception {
            System.out.println("**********:");
            ctx.writeAndFlush(new Message((byte)flag,System.currentTimeMillis(),scanner.nextLine()));
            return null;
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client inactive:"+new Date());
        //取消任务的执行
        this.futureTask.cancel(true);
    }

}

重连机制实现.png

二.服务端

首先是服务端的代码HeartServer

public class HeartServer {

    public static void main(String[] args){
        HeartServer server = new HeartServer();
        server.init(8080);
    }

    public void init (int port){

        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);

        NioEventLoopGroup workerGroup = new NioEventLoopGroup(10);

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        AcceptorIdleStateTrigger idleStateTrigger = new AcceptorIdleStateTrigger();
        serverBootstrap.option(ChannelOption.SO_BACKLOG, 128).
                group(bossGroup,workerGroup).
                channel(NioServerSocketChannel.class).
                childOption(ChannelOption.SO_KEEPALIVE, true).
                childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
                pipeline.addLast(new IdleStateHandler(5,0,0));
                pipeline.addLast(new MessageDecoder(Integer.MAX_VALUE,9,4,0,0,false));
                pipeline.addLast(new MessageEncoder());
                pipeline.addLast(idleStateTrigger);
                pipeline.addLast(new HeartServerHandler());
            }
        });

        try {
            ChannelFuture future = serverBootstrap.bind(port).sync();
            System.out.println("server start ...");
            //服务器同步连接断开时,这句代码才会往下执行
            future.channel().closeFuture().sync();
            System.out.println("server stop ...");
        } catch (InterruptedException e) {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            e.printStackTrace();
        }

    }
}

HeartServerHandler代码

public class HeartServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Server Active");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        Message message= (Message) msg;
        if (message.getType()==(byte)Integer.parseInt("CF", 16)){
            System.out.println(ctx.channel().remoteAddress()+">>>"+message.getType()+":"+message.getRequestId()+":"+message.getBody());
            ctx.writeAndFlush(new Message(message.getType(),message.getRequestId(),"server time->"+new Date()));
        }else if (message.getType()==(byte)Integer.parseInt("AF", 16)){
            System.out.println(ctx.channel().remoteAddress()+">>>>>>>>>>>>>>>>"+message.getBody());
        }
        ctx.fireChannelRead(msg);
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("已断开客户端"+ctx.channel().remoteAddress());
        ctx.channel().close();
    }
}

这里如果长时间没有触发读操作，就会自动和客户端断开连接，这种实现是基于IdleStateHandler来实现了，具体可以见我的上一篇文章。

@ChannelHandler.Sharable
public class AcceptorIdleStateTrigger extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent){
            IdleStateEvent stateEvent = (IdleStateEvent) evt;
            IdleState state = stateEvent.state();
            if (state==IdleState.READER_IDLE){
                throw new Exception("客户端空闲时间长，自动断开连接");
            }
        }else {
            super.userEventTriggered(ctx,evt);
        }
    }
}

三.自定义的拆包粘包规则

Message

public class Message {
    //消息类型
    private byte type;
    //请求ID
    private long requestId;
    //长度
    private int messageLength;
    //请求体
    private String body;

    public Message(byte type, long requestId, String body) {
        this.type = type;
        this.requestId = requestId;
        this.messageLength = body.getBytes(Charset.forName("UTF-8")).length;
        this.body = body;
    }

    public Message(byte type, long requestId, byte[] data) {
        this.type = type;
        this.requestId = requestId;
        this.messageLength = data.length;
        this.body = new String(data, Charset.forName("UTF-8"));
    }

    public byte getType() {
        return type;
    }

    public void setType(byte type) {
        this.type = type;
    }

    public long getRequestId() {
        return requestId;
    }

    public void setRequestId(long requestId) {
        this.requestId = requestId;
    }

    public int getMessageLength() {
        return messageLength;
    }

    public void setMessageLength(int messageLength) {
        this.messageLength = messageLength;
    }

    public String getBody() {
        return body;
    }

    public void setBody(String body) {
        this.body = body;
    }
}

MessageDecoder

public class MessageDecoder extends LengthFieldBasedFrameDecoder {

    private final byte HEADER_SIZE=13;

    public MessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip, boolean failFast) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, failFast);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        if (in == null || in.readableBytes()<HEADER_SIZE)return null;
        Object res = super.decode(ctx, in);
        ByteBuf processed = null;
        if (res!=null && res instanceof ByteBuf){
             processed = (ByteBuf) res;
        }else {
            throw new RuntimeException("协议异常");
        }
        byte type = processed.readByte();
        long requestId = processed.readLong();
        int length = processed.readInt();

        if (processed.readableBytes()<length){
            in.resetReaderIndex();
            return null;
        }
        byte[] data = new byte[length];
        processed.readBytes(data);
        return new Message(type,requestId,data);
    }
}

MessageEncoder

public class MessageEncoder extends MessageToByteEncoder<Message> {
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf byteBuf) throws Exception {
        byteBuf.writeByte(message.getType());
        byteBuf.writeLong(message.getRequestId());
        byte[] data = message.getBody().getBytes();
        byteBuf.writeInt(data.length);
        byteBuf.writeBytes(data);
    }
}