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

1. 创建SpringBoot父工程

创建一个springBoot工程，然后创建三个子模块
整体工程目录：一个server服务（Netty服务器），两个client服务（netty客户端）

pom文件引入netty依赖，springboot依赖

<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.0modelVersion>    <packaging>pompackaging>    <modules>        <module>servermodule>        <module>client1module>        <module>client2module>    modules>        <parent>        <groupId>org.springframework.bootgroupId>        <artifactId>spring-boot-starter-parentartifactId>        <version>2.2.1.RELEASEversion>        <relativePath/>     parent>    <groupId>org.examplegroupId>    <artifactId>nettyTestartifactId>    <version>1.0-SNAPSHOTversion>    <build>        <plugins>            <plugin>                <groupId>org.apache.maven.pluginsgroupId>                <artifactId>maven-compiler-pluginartifactId>                <configuration>                    <source>8source>                    <target>8target>                configuration>            plugin>        plugins>    build>    <dependencies>                <dependency>            <groupId>org.springframework.bootgroupId>            <artifactId>spring-boot-starter-WEBartifactId>        dependency>        <dependency>            <groupId>io.nettygroupId>            <artifactId>netty-allartifactId>            <version>4.1.34.Finalversion>        dependency>        <dependency>            <groupId>org.projectlombokgroupId>            <artifactId>lombokartifactId>            <version>1.16.18version>        dependency>        <dependency>            <groupId>org.slf4jgroupId>            <artifactId>slf4j-apiartifactId>        dependency>        <dependency>            <groupId>com.alibabagroupId>            <artifactId>fastJSONartifactId>            <version>1.1.23version>        dependency>    dependencies>project>

2. 创建Netty服务器

NettySpringBootApplication

package server;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic 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;@Componentpublic 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@Slf4jpublic class NettyServer {    private final static Logger logger = LoggerFactory.getLogger(NettyServer.class);    @Resource    private SocketInitializer socketInitializer;    @Getter    private ServerBootstrap serverBootstrap;        @Value("${netty.port:6666}")    private int port;        @Value("${netty.bossThread:1}")    private int bossThread;        public void start() {        this.init();        this.serverBootstrap.bind(this.port);        logger.info("Netty started on port: {} (tcp) with boss thread {}", this.port, this.bossThread);    }        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;@Componentpublic class NettyStartListener implements ApplicationRunner {    @Resource    private NettyServer nettyServer;    @Override    public void run(ApplicationArguments args) throws Exception {        this.nettyServer.start();    }}

application.yml

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

3. 创建Netty客户端

Client1

package client;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic 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;@Componentpublic 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@Slf4jpublic class NettyClient {    private final static Logger logger = LoggerFactory.getLogger(NettyClient.class);    @Resource    private SocketInitializer socketInitializer;    @Getter    private Bootstrap bootstrap;    @Getter    private Channel channel;        @Value("${netty.port:6666}")    private int port;    @Value("${netty.host:127.0.0.1}")    private String host;        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();        }    }        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;@Componentpublic class NettyStartListener implements ApplicationRunner {    @Resource    private NettyClient nettyClient;    @Override    public void run(ApplicationArguments args) throws Exception {        this.nettyClient.start();    }}

application.yml

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

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

4. 测试

来源地址：https://blog.csdn.net/weixin_42774617/article/details/131524243