netty学习（3）:SpringBoot整合netty实现多个客户端与服务器通信

亦凉 2023-10-13 10:45 33阅读 0赞

### 1. 创建SpringBoot父工程 ###

创建一个SpringBoot工程，然后创建三个子模块  
整体工程目录：一个server服务（netty服务器），两个client服务（netty客户端）  
![在这里插入图片描述][351194b7193348078852ebc250eaad86.png_pic_center]  
pom文件引入netty依赖，springboot依赖

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
        <packaging>pom</packaging>
        <modules>
            <module>server</module>
            <module>client1</module>
            <module>client2</module>
        </modules>
    
        
        <parent>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>2.2.1.RELEASE</version>
            <relativePath/> 
        </parent>
    
        <groupId>org.example</groupId>
        <artifactId>nettyTest</artifactId>
        <version>1.0-SNAPSHOT</version>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>8</source>
                        <target>8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
        <dependencies>
            
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.34.Final</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.16.18</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.1.23</version>
            </dependency>
        </dependencies>
    </project>

### 2. 创建Netty服务器 ###

![在这里插入图片描述][cd70bc3013a04ff682b6a752dc0e33cb.png]  
NettySpringBootApplication

package server;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class NettySpringBootApplication {
        
        public static void main(String[] args) {
        
            SpringApplication.run(NettySpringBootApplication.class, args);
        }
    }

NettyServiceHandler

package server.netty;
    
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    public class NettyServiceHandler extends SimpleChannelInboundHandler<String> {
        
        private static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            // 获取到当前与服务器连接成功的channel
            Channel channel = ctx.channel();
            group.add(channel);
            System.out.println(channel.remoteAddress() + " 上线," + "在线数量：" + group.size());
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        
            // 获取到当前要断开连接的Channel
            Channel channel = ctx.channel();
            System.out.println(channel.remoteAddress() + "下线，" + "在线数量：" + group.size());
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        
            Channel channel = ctx.channel();
            System.out.println("netty客户端" + channel.remoteAddress() + "发送过来的消息:" + msg);
            group.forEach(ch -> {
         // JDK8 提供的lambda表达式
                if (ch != channel) {
        
                    ch.writeAndFlush(channel.remoteAddress() + "：" + msg + "\n");
                }
            });
        }
    
        public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
        
            throwable.printStackTrace();
            channelHandlerContext.close();
        }
    }

SocketInitializer

package server.netty;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.bytes.ByteArrayDecoder;
    import io.netty.handler.codec.bytes.ByteArrayEncoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import org.springframework.stereotype.Component;
    
    @Component
    public class SocketInitializer extends ChannelInitializer<SocketChannel> {
        
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
        
            ChannelPipeline pipeline = socketChannel.pipeline();
    //        // 添加对byte数组的编解码，netty提供了很多编解码器，你们可以根据需要选择
    //        pipeline.addLast(new ByteArrayDecoder());
    //        pipeline.addLast(new ByteArrayEncoder());
    
            //添加一个基于行的解码器
            pipeline.addLast(new LineBasedFrameDecoder(2048));
            pipeline.addLast(new StringDecoder());
            pipeline.addLast(new StringEncoder());
    
            // 添加上自己的处理器
            pipeline.addLast(new NettyServiceHandler());
        }
    }

NettyServer

package server.netty;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import lombok.Getter;
    import lombok.extern.slf4j.Slf4j;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    @Component
    @Slf4j
    public class NettyServer {
        
        private final static Logger logger = LoggerFactory.getLogger(NettyServer.class);
    
        @Resource
        private SocketInitializer socketInitializer;
    
        @Getter
        private ServerBootstrap serverBootstrap;
    
        /**
         * netty服务监听端口
         */
        @Value("${netty.port:6666}")
        private int port;
        /**
         * 主线程组数量
         */
        @Value("${netty.bossThread:1}")
        private int bossThread;
    
        /**
         * 启动netty服务器
         */
        public void start() {
        
            this.init();
            this.serverBootstrap.bind(this.port);
            logger.info("Netty started on port: {} (TCP) with boss thread {}", this.port, this.bossThread);
        }
    
        /**
         * 初始化netty配置
         */
        private void init() {
        
            NioEventLoopGroup bossGroup = new NioEventLoopGroup(this.bossThread);
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            this.serverBootstrap = new ServerBootstrap();
            this.serverBootstrap.group(bossGroup, workerGroup) // 两个线程组加入进来
                    .channel(NioServerSocketChannel.class)  // 配置为nio类型
                    .option(ChannelOption.SO_BACKLOG, 128) //设置线程队列等待连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(this.socketInitializer); // 加入自己的初始化器
        }
    }

NettyStartListener

package server.netty;
    
    import org.springframework.boot.ApplicationArguments;
    import org.springframework.boot.ApplicationRunner;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    /**
     * 监听Spring容器启动完成，完成后启动Netty服务器
     * @author Gjing
     **/
    @Component
    public class NettyStartListener implements ApplicationRunner {
        
        @Resource
        private NettyServer nettyServer;
    
        @Override
        public void run(ApplicationArguments args) throws Exception {
        
            this.nettyServer.start();
        }
    }

application.yml

server:
      port: 8000
    
    netty:
      port: 6666
      bossThread: 1

### 3. 创建Netty客户端 ###

![在这里插入图片描述][6e35559a111843fcbb4403322f712fa7.png]  
Client1

package client;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class Client1 {
        
        public static void main(String[] args) {
        
            SpringApplication.run(Client1.class, args);
        }
    }

NettyClientHandler

package client.netty;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    public class NettyClientHandler extends SimpleChannelInboundHandler<String> {
        
        public void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) {
        
            System.out.println("收到服务端消息：" + channelHandlerContext.channel().remoteAddress() + "的消息：" + msg);
        }
    
        public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
        
            channelHandlerContext.close();
        }
    }

SocketInitializer

package client.netty;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import org.springframework.stereotype.Component;
    
    @Component
    public class SocketInitializer extends ChannelInitializer<SocketChannel> {
        
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
        
            ChannelPipeline pipeline = socketChannel.pipeline();
            pipeline.addLast(new LineBasedFrameDecoder(2048));
            pipeline.addLast(new StringDecoder());
            pipeline.addLast(new StringEncoder());
            pipeline.addLast(new NettyClientHandler()); //加入自己的处理器
        }
    }

NettyClient

package client.netty;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import lombok.Getter;
    import lombok.extern.slf4j.Slf4j;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    import javax.annotation.Resource;
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.nio.charset.StandardCharsets;
    
    @Component
    @Slf4j
    public class NettyClient {
        
        private final static Logger logger = LoggerFactory.getLogger(NettyClient.class);
    
        @Resource
        private SocketInitializer socketInitializer;
    
        @Getter
        private Bootstrap bootstrap;
    
        @Getter
        private Channel channel;
        /**
         * netty服务监听端口
         */
        @Value("${netty.port:6666}")
        private int port;
    
        @Value("${netty.host:127.0.0.1}")
        private String host;
    
        /**
         * 启动netty
         */
        public void start() {
        
            this.init();
            this.channel = this.bootstrap.connect(host, port).channel();
            logger.info("Netty connect on port: {}, the host {}, the channel {}", this.port, this.host, this.channel);
            try {
        
                InputStreamReader is = new InputStreamReader(System.in, StandardCharsets.UTF_8);
                BufferedReader br = new BufferedReader(is);
                while (true) {
        
                    System.out.println("输入：");
                    this.channel.writeAndFlush(br.readLine() + "\r\n");
                }
            } catch (IOException e) {
        
                e.printStackTrace();
            }
        }
    
        /**
         * 初始化netty配置
         */
        private void init() {
        
            EventLoopGroup group = new NioEventLoopGroup();
            this.bootstrap = new Bootstrap();
            //设置线程组
            bootstrap.group(group)
                    .channel(NioSocketChannel.class) //设置客户端的通道实现类型
                    .handler(this.socketInitializer);
        }
    }

NettyStartListener

package client.netty;
    
    import org.springframework.boot.ApplicationArguments;
    import org.springframework.boot.ApplicationRunner;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    @Component
    public class NettyStartListener implements ApplicationRunner {
        
        @Resource
        private NettyClient nettyClient;
    
        @Override
        public void run(ApplicationArguments args) throws Exception {
        
            this.nettyClient.start();
        }
    }

application.yml

server:
      port: 8001
    
    netty:
      port: 6666
      host: "127.0.0.1"

然后按照相同的方法创建client2

### 4. 测试 ###

![在这里插入图片描述][92ebe1f30f4840a59535243c832ace5c.png_pic_center]  
![在这里插入图片描述][ce7dfc312bd44506b6a0c5d2557d9ef0.png_pic_center]

[351194b7193348078852ebc250eaad86.png_pic_center]: https://img-blog.csdnimg.cn/351194b7193348078852ebc250eaad86.png#pic_center
[cd70bc3013a04ff682b6a752dc0e33cb.png]: https://img-blog.csdnimg.cn/cd70bc3013a04ff682b6a752dc0e33cb.png
[6e35559a111843fcbb4403322f712fa7.png]: https://img-blog.csdnimg.cn/6e35559a111843fcbb4403322f712fa7.png
[92ebe1f30f4840a59535243c832ace5c.png_pic_center]: https://img-blog.csdnimg.cn/92ebe1f30f4840a59535243c832ace5c.png#pic_center
[ce7dfc312bd44506b6a0c5d2557d9ef0.png_pic_center]: https://img-blog.csdnimg.cn/ce7dfc312bd44506b6a0c5d2557d9ef0.png#pic_center