netty重连

101 阅读 0 评论 67 点赞

我是靠谱客的博主玩命画笔，最近开发中收集的这篇文章主要介绍netty重连，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

简单的netty重连
1. 首先是有一个 netty 客户端；

`package com.example.nettySimple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

import java.util.concurrent.TimeUnit;

/**
 * @author Xia_Ye_bing
 * @PRODUCT_NAME IntelliJ IDEA
 * @creatime 2021 10 27 13:20
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client" + ctx); 
    }

    /**
     * 当通道有读取事件时 会触发，
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        final ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务器回复的信息是 " + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器地址为：" + ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();
        System.out.println(" netty 链接失败，3秒后进行重连");
        Thread.sleep(3000);
        NettyClient.start();
    }

    @Override
    public void channelInactive(final ChannelHandlerContext ctx) {
        System.out.println("连接断开:" + ctx.channel().remoteAddress());
    }

}

`创建一个处理器

  package com.example.nettySimple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

import java.util.concurrent.TimeUnit;

/**
 * @author Xia_Ye_bing
 * @PRODUCT_NAME IntelliJ IDEA
 * @creatime 2021 10 27 13:20
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client" + ctx);
    }

    /**
     * 当通道有读取事件时 会触发，
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        final ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务器回复的信息是 " + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器地址为：" + ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();
        System.out.println(" netty 链接失败，3秒后进行重连");
        Thread.sleep(3000);
        NettyClient.start();
    }

    @Override
    public void channelInactive(final ChannelHandlerContext ctx) {
        System.out.println("连接断开:" + ctx.channel().remoteAddress());
    }

}

演示效果
netty重连演示-url（需下载）
新建一个客户端

package com.example.nettySimple;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * @author Xia_Ye_bing
 * @PRODUCT_NAME IntelliJ IDEA
 * @creatime 2021 10 27 7:19
 */
public class nrtty_siple {

    public static void main(String[] args) throws InterruptedException {
        /**
         * 创建两个 线程组 booGroup 和 workerGroup，
         * booGroup 只处理链接请求，
         * workerGroup 真正处理客户端业务处理
         * 两个都是 无线循环，两个线程的默认线程个数是 CPU核数*2，写了值话，按照写的个数创建线程数
         */
        final EventLoopGroup booGroup = new NioEventLoopGroup(1);

        final EventLoopGroup workerGroup = new NioEventLoopGroup();

        /**
         *    创建启动对象
         */
        final ServerBootstrap bootstrap = new ServerBootstrap();

        /**
         * 使用链式编程 来 设置 各个参数
         */

        bootstrap.group(booGroup, workerGroup)//设置里昂个线程组
                .channel(NioServerSocketChannel.class)//使用  NioServerSocketChannel 作为服务器的通道实现
                .option(ChannelOption.SO_BACKLOG, 128)//设置保持通道链接状态
                .childOption(ChannelOption.SO_KEEPALIVE, true)//对应的管道设置处理器
                .childHandler(new ChannelInitializer<SocketChannel>() {//创建一个通道初始化对象（匿名对象）
                    //给 pipeline 设置处理器
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new nettySipleHandler());//将自定义的处理器加入
                    }
                });//给我们的workerGroup 的 EventLoopGroup 设置处理器
        System.out.println("服务器 is ready");
        //绑定一个端口 生成一个channelFuture 对象 同时启动服务器
        final ChannelFuture channelFuture = bootstrap.bind(6668).sync();
        //对关闭通道进行监听
        channelFuture.channel().closeFuture().sync();

    }
}

8 建立客户端处理器

package com.example.nettySimple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

/**
 * @author Xia_Ye_bing
 * @PRODUCT_NAME IntelliJ IDEA
 * @creatime 2021 10 27 7:40
 */

/**
 * 我，们自定义一个handler ，需要 继承 netty规定好的 某个 HandlerAdapter （规范）
 * 这时的handler 才是 一个 handler
 */
public class nettySipleHandler extends ChannelInboundHandlerAdapter {
    //读取数据事件 （）可以读取客户端发送的消息

    /**
     * @param ctx ： ChannelHandlerContext ctx 上下文对象，含有 通道 、管道
     *            Object ms 客户端发发送的数据
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx" + msg);
        System.out.println("服务器读取线程   "+Thread.currentThread().getName());
        /**
         * 将msg转为bytebuf ，
         * bytebuf  是 netty 提供，不是 nio中的ByteBuffer
         */
        final ByteBuf buf = (ByteBuf) msg;
        System.out.println("客户端发送的消息是 " + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址是：" + ctx.channel().remoteAddress());
    }

    //数据读取完毕
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

        /**
         * writeAndFlush 写并且发送
         * 将数据写入缓存并刷新
         * 需要对发送的数据进行编码
         */
        ctx.writeAndFlush(Unpooled.copiedBuffer(("hellow 客户端"), CharsetUtil.UTF_8));
    }

    //关闭通道、
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().close();
    }
}