springboot集成netty服务端(TCP socket服务端)

末蓝、 2023-10-10 14:26 61阅读 0赞

**netty依赖:**

<dependency>
    		<groupId>io.netty</groupId>
    		<artifactId>netty-all</artifactId>
    		<version>4.1.28.Final</version>
    	</dependency>

**代码:**

**server启动类**

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    import java.net.InetSocketAddress;
    
    /**
     * @description: netty服务启动类
     **/
    
    @Slf4j
    @Component
    public class NettyServer {
        
    
        public void start(InetSocketAddress address) {
        
            //配置服务端的NIO线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try {
        
                ServerBootstrap bootstrap = new ServerBootstrap()
                        .group(bossGroup, workerGroup)  // 绑定线程池
                        .channel(NioServerSocketChannel.class)
                        .localAddress(address)
                        .childHandler(new NettyServerChannelInitializer())//编码解码
                        .option(ChannelOption.SO_BACKLOG, 128)  //服务端接受连接的队列长度，如果队列已满，客户端连接将被拒绝
                        .childOption(ChannelOption.SO_KEEPALIVE, true);  //保持长连接，2小时无数据激活心跳机制
    
                // 绑定端口，开始接收进来的连接
                ChannelFuture future = bootstrap.bind(address).sync();
                log.info("netty服务器开始监听端口：" + address.getPort());
                //关闭channel和块，直到它被关闭
                future.channel().closeFuture().sync();
            } catch (Exception e) {
        
                e.printStackTrace();
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
    }

**server初始化类**

import io.netty.channel.ChannelInitializer;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.util.CharsetUtil;
    
    /**
     * @description: 服务端初始化，客户端与服务器端连接一旦创建，这个类中方法就会被回调，设置出站编码器和入站解码器
     **/
    
    public class NettyServerChannelInitializer extends ChannelInitializer<SocketChannel> {
        
    
        @Override
        protected void initChannel(SocketChannel channel) throws Exception {
        
    
            channel.pipeline().addLast("decoder",new StringDecoder(CharsetUtil.UTF_8));
            channel.pipeline().addLast("encoder",new StringEncoder(CharsetUtil.UTF_8));
            channel.pipeline().addLast(new NettyServerHandler());
        }
    }

**server处理类**

import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelId;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    import java.net.InetSocketAddress;
    import java.util.concurrent.ConcurrentHashMap;
    
    /**
     * @description: netty服务端处理类
     **/
    
    @Slf4j
    @Component
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
        
    
        /**
         * 管理一个全局map，保存连接进服务端的通道数量
         */
        private static final ConcurrentHashMap<ChannelId, ChannelHandlerContext> CHANNEL_MAP = new ConcurrentHashMap<>();
    
       
    
        /**
         * @param ctx
         * @DESCRIPTION: 有客户端连接服务器会触发此函数
         * @return: void
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
        
    
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
    
            String clientIp = insocket.getAddress().getHostAddress();
            int clientPort = insocket.getPort();
    
            //获取连接通道唯一标识
            ChannelId channelId = ctx.channel().id();
    
            System.out.println();
            //如果map中不包含此连接，就保存连接
            if (CHANNEL_MAP.containsKey(channelId)) {
        
                log.info("客户端【" + channelId + "】是连接状态，连接通道数量: " + CHANNEL_MAP.size());
            } else {
        
                //保存连接
                CHANNEL_MAP.put(channelId, ctx);
    
                log.info("客户端【" + channelId + "】连接netty服务器[IP:" + clientIp + "--->PORT:" + clientPort + "]");
                log.info("连接通道数量: " + CHANNEL_MAP.size());
            }
        }
    
        /**
         * @param ctx
         * @DESCRIPTION: 有客户端终止连接服务器会触发此函数
         * @return: void
         */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
        
    
            InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
    
            String clientIp = insocket.getAddress().getHostAddress();
    
            ChannelId channelId = ctx.channel().id();
    
            //包含此客户端才去删除
            if (CHANNEL_MAP.containsKey(channelId)) {
        
                //删除连接
                CHANNEL_MAP.remove(channelId);
                System.out.println();
                log.info("客户端【" + channelId + "】退出netty服务器[IP:" + clientIp + "--->PORT:" + insocket.getPort() + "]");
                log.info("连接通道数量: " + CHANNEL_MAP.size());
            }
        }
    
        /**
         * @param ctx
         * @DESCRIPTION: 有客户端发消息会触发此函数
         * @return: void
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
    
            String msgStr = String.valueOf(msg);
    
            System.out.println();
            log.info("加载客户端报文......");
            log.info("【" + ctx.channel().id() + "】" + " :" + msg);
    
       
            /**
             *  下面可以解析数据，保存数据，生成返回报文，将需要返回报文写入write函数
             *
             */
    
            //响应客户端
            log.info("服务端端返回报文......");
            log.info("【" + ctx.channel().id() + "】" + " :" + msg);
            this.channelWrite(ctx.channel().id(), msg);
        }
    
        /**
         * @param msg        需要发送的消息内容
         * @param channelId 连接通道唯一id
         * @DESCRIPTION: 服务端给客户端发送消息
         * @return: void
         */
        public void channelWrite(ChannelId channelId, Object msg) throws Exception {
        
    
            ChannelHandlerContext ctx = CHANNEL_MAP.get(channelId);
    
            if (ctx == null) {
        
                log.info("通道【" + channelId + "】不存在");
                return;
            }
    
            if (msg == null || msg == "") {
        
                log.info("服务端响应空的消息");
                return;
            }
    
            //将客户端的信息直接返回写入ctx
            ctx.write(msg);
            //刷新缓存区
            ctx.flush();
        }
    
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        
    
            String socketString = ctx.channel().remoteAddress().toString();
    
            if (evt instanceof IdleStateEvent) {
        
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state() == IdleState.READER_IDLE) {
        
                    log.info("Client: " + socketString + " READER_IDLE 读超时");
                    ctx.disconnect();
                } else if (event.state() == IdleState.WRITER_IDLE) {
        
                    log.info("Client: " + socketString + " WRITER_IDLE 写超时");
                    ctx.disconnect();
                } else if (event.state() == IdleState.ALL_IDLE) {
        
                    log.info("Client: " + socketString + " ALL_IDLE 总超时");
                    ctx.disconnect();
                }
            }
        }
    
        /**
         * @param ctx
         * @DESCRIPTION: 发生异常会触发此函数
         * @return: void
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
    
            System.out.println();
            ctx.close();
            log.info(ctx.channel().id() + " 发生了错误,此连接被关闭" + "此时连通数量: " + CHANNEL_MAP.size());
            //cause.printStackTrace();
        }
    }

**启动类**

> 在springboot启动类中加入服务端的启动代码，让netty服务跟着项目一起启动。

@Slf4j
    @SpringBootApplication
    @EnableAutoConfiguration(exclude={
        MongoAutoConfiguration.class})
    public class JeecgSystemApplication extends SpringBootServletInitializer implements CommandLineRunner{
        
    
        @Value("${netty.port}")
        private Integer port;
        @Value("${netty.host}")
        private String host;
    
        @Autowired
        private NettyServer nettyServer;
    
        @Override
        protected SpringApplicationBuilder configure(SpringApplicationBuilder application) {
        
            return application.sources(JeecgSystemApplication.class);
        }
    
        public static void main(String[] args) throws UnknownHostException {
        
            ConfigurableApplicationContext application = SpringApplication.run(JeecgSystemApplication.class, args);
            Environment env = application.getEnvironment();
            String ip = InetAddress.getLocalHost().getHostAddress();
            String port = env.getProperty("server.port");
            String path = oConvertUtils.getString(env.getProperty("server.servlet.context-path"));
        }
    
        /**
         * netty服务启动
         * @param args
         * @throws Exception
         */
        @Override
        public void run(String... args) throws Exception {
        
            InetSocketAddress address = new InetSocketAddress(host,port);
            log.info("neety服务器启动地址: "+host+":"+ port);
            nettyServer.start(address);
        }
    }

### 结束! ###