netty-发起tcp长连接(包含客户端和服务端)

待我称王封你为后i 2023-10-04 19:01 175阅读 0赞

Netty是一个高性能、异步事件驱动的NIO框架，它提供了对TCP、UDP和文件传输的支持。

Netty是对JDK自带的NIO的API进行封装，具有高并发，高性能等优点。

项目中经常用到netty实现服务器与设备的通信，先写服务端代码：

@Slf4j
    @Component
    public class NettyServerBootstrap {
    
        private Channel serverChannel;
        private static final int DEFAULT_PORT = 60782;
        //bossGroup只是处理连接请求
        private static EventLoopGroup bossGroup = null;
        //workGroup处理非连接请求，如果牵扯到数据量处理业务非常耗时的可以再单独新建一个eventLoopGroup,并在childHandler初始化的时候添加到pipeline绑定
        private static EventLoopGroup workGroup = null;
    
        /**
         * 启动Netty服务
         *
         * @return 启动结果
         */
        @PostConstruct
        public boolean start() {
            bossGroup = new NioEventLoopGroup();
            workGroup = new NioEventLoopGroup();
            //创建服务端启动对象
            ServerBootstrap bootstrap = new ServerBootstrap();
            try {
                //使用链式编程来设置
                bootstrap.group(bossGroup, workGroup)//设置两个线程组
                        //使用NioSocketChannel作为服务器的通道实现
                        .channel(NioServerSocketChannel.class)
                        //设置线程队列得到的连接数
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        //设置保持活动连接状态
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        //设置处理器  WorkerGroup 的 EvenLoop 对应的管道设置处理器
                        .childHandler(new ChannelInitializer<Channel>() {
    
                            @Override
                            protected void initChannel(Channel ch){
                                log.info("--------------有客户端连接");
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new StringEncoder());
                                ch.pipeline().addLast(new NettyServerHandler());
                            }
                        });
                //绑定端口, 同步等待成功;
                ChannelFuture future = bootstrap.bind(DEFAULT_PORT).sync();
                log.info("netty服务启动成功，ip：{}，端口：{}", InetAddress.getLocalHost().getHostAddress(), DEFAULT_PORT);
                serverChannel = future.channel();
                ThreadUtil.execute(() -> {
                    //等待服务端监听端口关闭
                    try {
                        future.channel().closeFuture().sync();
                        log.info("netty服务正常关闭成功，ip：{}，端口：{}", InetAddress.getLocalHost().getHostAddress(), DEFAULT_PORT);
                    } catch (InterruptedException | UnknownHostException e) {
                        e.printStackTrace();
                    } finally {
                        shutdown();
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
                log.error("netty服务异常，异常原因：{}", e.getMessage());
                return false;
            }
            return true;
        }
    
    
        /**
         * 关闭当前server
         */
        public boolean close() {
            if (serverChannel != null) {
                serverChannel.close();//关闭服务
                try {
                    //保险起见
                    serverChannel.closeFuture().sync();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return false;
                } finally {
                    shutdown();
                    serverChannel = null;
                }
            }
            return true;
        }
    
        /**
         * 优雅关闭
         */
        private void shutdown() {
            workGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    
    }

服务端处理类代码：

@Slf4j
    public class NettyServerHandler extends SimpleChannelInboundHandler<String> {
    
        /**
         * 处理读取到的msg
         *
         * @param ctx 上下文
         * @param msg 数据
         */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx,String msg) throws Exception {
            System.out.println("服务端收到的消息--------"+msg);
            ctx.channel().writeAndFlush("ok");
        }
    
        /**
         * 断开连接
         *
         * @param ctx 傻瓜下文
         */
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            ChannelId channelId = ctx.channel().id();
            log.info("客户端id:{},断开连接,ip:{}", channelId, ctx.channel().remoteAddress());
            super.handlerRemoved(ctx);
        }
    
    }

接下来模拟客户端：

@Configuration
    @Component
    public class TianmiaoClient {
    
    
        private static String ip;
    
        private static int port ;
    
        @Value("${tianmiao.nettyServer.ip}")
        public void setIp(String ip) {
            this.ip = ip;
        }
    
        @Value("${tianmiao.nettyServer.port}")
        public void setPort(int port) {
            this.port = port;
        }
    
        /**
         * 服务类
         */
        private static Bootstrap bootstrap=null;
    
        /**
         * 初始化  项目启动后自动初始化
         */
        @PostConstruct
        public void init() {
    
            //worker
            EventLoopGroup worker = new NioEventLoopGroup();
    
            bootstrap = new Bootstrap();
            //设置线程池
            bootstrap.group(worker);
    
            //设置socket工厂
            bootstrap.channel(NioSocketChannel.class);
    
            //设置管道
            bootstrap.handler(new ChannelInitializer<Channel>() {
    
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new StringDecoder());
                    ch.pipeline().addLast(new StringEncoder());
                    ch.pipeline().addLast(new TianmiaoClientHandler());
                }
            });
        }
    
    
        /**
         * 获取会话 （获取或者创建一个会话）
         */
        public Channel createChannel() {
            try {
                Channel channel = bootstrap.connect( ip, port).sync().channel();
                return channel;
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }
    
    }

客户端处理类代码

public class TianmiaoClientHandler extends SimpleChannelInboundHandler<String> {
    
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
            System.out.println("服务端发过来的消息："+s);
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            System.out.println(".......................tcp断开连接.........................");
            //移除
            Channel channel = ctx.channel();
            channel.close().sync();
            super.channelInactive(ctx);
        }
    
    }

管理客户端channel的一个工具类：

public class TianmiaoChannelManager {
    
        /**
         * 在线会话(存储注册成功的会话)
         */
        private static final ConcurrentHashMap<String, Channel> onlineChannels = new ConcurrentHashMap<>();
    
    
        /**
         * 加入
         *
         * @param mn
         * @param channel
         * @return
         */
        public static boolean putChannel(String mn, Channel channel) {
            if (!onlineChannels.containsKey(mn)) {
                boolean success = onlineChannels.putIfAbsent(mn, channel) == null ? true : false;
                return success;
            }
            return false;
        }
    
        /**
         * 移除
         *
         * @param mn
         */
        public static Channel removeChannel(String mn) {
            return onlineChannels.remove(mn);
        }
    
        /**
         * 获取Channel
         *
         * @param mn
         * @return
         */
        public static Channel getChannel(String mn) {
            // 获取一个可用的会话
            Channel channel = onlineChannels.get(mn);
            if (channel != null) {
                // 连接有可能是断开，加入已经断开连接了，我们需要进行尝试重连
                if (!channel.isActive()) {
                    //先移除之前的连接
                    removeChannel(mn);
                    return null;
                }
            }
            return channel;
        }
    
        /**
         * 发送消息[自定义协议]
         *
         * @param <T>
         * @param mn
         * @param msg
         */
        public static <T> void sendMessage(String mn, String msg) {
            Channel channel = onlineChannels.get(mn);
            if (channel != null && channel.isActive()) {
                channel.writeAndFlush(msg);
            }
        }
    
        /**
         * 发送消息[自定义协议]
         *
         * @param <T>
         * @param msg
         */
        public static <T> void sendChannelMessage(Channel channel, String msg) {
            if (channel != null && channel.isActive()) {
                channel.writeAndFlush(msg);
            }
        }
    
        /**
         * 关闭连接
         *
         * @return
         */
        public static void closeChannel(String mn) {
            onlineChannels.get(mn).close();
        }
    }

最后是客户端使用方法：

/**
     * 发送数据包
     * @param key
     */
    public static void tianmiaoData(String key, String data) {
        Channel channel = TianmiaoChannelManager.getChannel(key);
        //将通道存入
        if(channel==null){
            TianmiaoClient client = new TianmiaoClient();
            channel = client.createChannel();
            TianmiaoChannelManager.putChannel(key, channel);
        }
        if (channel != null && channel.isActive()) {
            //发送数据
            channel.writeAndFlush(data);
            System.out.println("-------------天苗转发数据成功-------------");
        }
    }