netty 客户端 实现断开重连

1、首先引入依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.6.Final</version>
</dependency>
1
2
3
4
5

2、创建server层代码

2.1、编写服务端代码

public static void main(String[] args) {
	new Thread(()->{
	    NioEventLoopGroup bossGroup = new NioEventLoopGroup();
	    NioEventLoopGroup workGroup = new NioEventLoopGroup();
	    try {
	        ServerBootstrap bootstrap = new ServerBootstrap();
	        bootstrap.group(bossGroup,workGroup)
	                .channel(NioServerSocketChannel.class)
	                .option(ChannelOption.SO_BACKLOG,128)
	                .childOption(ChannelOption.SO_KEEPALIVE,true)
	                .childHandler(new ServerChannelInitializer());
	        ChannelFuture channelFuture = bootstrap.bind(8099).sync();
	        channelFuture.channel().closeFuture().sync();
	    }catch (Exception e){
	    }finally {
	        bossGroup.shutdownGracefully();
	        workGroup.shutdownGracefully();
	    }
	}).start();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2.2、创建childHandler处理接收handler

public static class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new IdleStateHandler(10,10,10, TimeUnit.SECONDS));
        pipeline.addLast(new ReadHandler());
        pipeline.addLast(new WriteHandler());
        pipeline.addLast(new ServerSendHandler());
    }
}
1
2
3
4
5
6
7
8
9
10

2.3、创建公共的读写处理器

public static class ReadHandler extends SimpleChannelInboundHandler<ByteBuf>{
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        System.out.println("from : "+ctx.channel().remoteAddress()+" data : "+ ByteBufUtil.hexDump(msg));
        msg.retain();
        ctx.fireChannelRead(msg);
    }
}

public static class WriteHandler extends ChannelOutboundHandlerAdapter{
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (msg instanceof ByteBuf){
            ByteBuf buf = (ByteBuf)msg;
            System.out.println("to : "+ctx.channel().remoteAddress()+" data : "+ByteBufUtil.hexDump(buf));
            super.write(ctx, msg, promise);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2.4、创建服务端的简单业务处理器

public static class ServerSendHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        private static AtomicInteger atomicInteger = new AtomicInteger(1);

    	//这里简单模拟消息发送
        int a = msg.readInt();
        long b = msg.readLong();
        System.out.println("server ---- a:"+a +"  b:"+b);

        ByteBuf buffer = Unpooled.buffer();
        buffer.writeInt(atomicInteger.getAndIncrement());
        buffer.writeLong(170);
        ctx.writeAndFlush(buffer);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        System.out.println("server disconnect "+ctx.channel().remoteAddress());
        ctx.channel().disconnect();
        ctx.fireChannelInactive();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                // 读空闲超时，断开连接
                System.out.println("Channel read timeout, remote address:" + ctx.channel().remoteAddress().toString());
                ctx.close();
            } else if (state == IdleState.WRITER_IDLE) {
                ByteBuf buffer = Unpooled.buffer();
                buffer.writeInt(101);
                buffer.writeLong(171);
                ctx.writeAndFlush(buffer);
                System.out.println("Channel write timeout, remote address:" + ctx.channel().remoteAddress().toString());

            }
        }

        super.userEventTriggered(ctx, evt);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

3、创建客户端代码连接

3.1、编写客户端代码

public static void main(String[] args) {
    new Thread(()->{
        client(new NioEventLoopGroup());
    }).start();
}
 public static void client(NioEventLoopGroup group){
     Bootstrap bootstrap = new Bootstrap();
     bootstrap.group(group)
             .channel(NioSocketChannel.class)
             .option(ChannelOption.SO_KEEPALIVE,true)
             .handler(new ClientChannelInitializer());
     try {
         bootstrap.connect("127.0.0.1",8099).addListener(new ClientChannelListener()).sync();
     } catch (InterruptedException e) {
         e.printStackTrace();
     }
 }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

2.2、创建handler处理接收handler

public static class ClientChannelInitializer extends ChannelInitializer<SocketChannel>{
  @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new IdleStateHandler(10,10,10, TimeUnit.SECONDS));
        pipeline.addLast(new ReadHandler());
        pipeline.addLast(new WriteHandler());
        pipeline.addLast(new ClientActHandler());
    }
}
1
2
3
4
5
6
7
8
9
10

3.3、创建客户端的简单业务处理器实现断线重连

public static class ClientActHandler extends SimpleChannelInboundHandler<ByteBuf>{
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        ByteBuf buffer = Unpooled.buffer();
        buffer.writeInt(99);
        buffer.writeLong(180);
        ctx.writeAndFlush(buffer);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        int a = msg.readInt();
        long b = msg.readLong();
        System.out.println("client : "+a+"  "+b);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        System.out.println("重新建立新的连接。。。。。");
        client(new NioEventLoopGroup());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

3.3、客户端监听器

public static class ClientChannelListener implements ChannelFutureListener{
    @Override
     public void operationComplete(ChannelFuture future) throws Exception {
         boolean success = future.isSuccess();
         if (success){
             System.out.println("connect success ");
         }else {
             System.out.println("connect error");
         }
     }
 }
1
2
3
4
5
6
7
8
9
10
11

4、结果输出

可以看到因为读超时服务端断开连接，然后客户端又重新连接

服务端

客户端