准备工作

如果没有现成的硬件设备可以通过下载网络调试助手进行模拟。

填写好远程主机地址，以及netty服务端口。

编写netty服务端

@Component
public class NettyServer {

    /**
     * 服务端口
     */
    @Value("${netty.port}")
    private int port;

    public static NioServerSocketChannel nioServerSocketChannel;

    @Autowired
    private ServerHandler serverHandler;

    public void start() {
        //连接处理group
        NioEventLoopGroup boss = new NioEventLoopGroup();
        //事件处理group
        NioEventLoopGroup worker = new NioEventLoopGroup();
        //创建serverBootStrap实例
        ServerBootstrap bootstrap = new ServerBootstrap();
        //1.绑定group
        bootstrap.group(boss, worker)
                //2.设置并绑定Reactor线程池：EventLoopGroup，EventLoop就是处理所有注册到本线程的Selector上面的Channel
                .channel(NioServerSocketChannel.class)
                //3.保存连接数
                .option(ChannelOption.SO_BACKLOG, 1024)
                //4.有数据立即发送
                .option(ChannelOption.SO_SNDBUF, 4 * 1024)
                //5.保持连接
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                //6.处理新连接
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    //设置了客户端连接socket属性。
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //增加任务处理
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(
                                //自定义解码器
                                new DecoderHandler()
                                , new StringEncoder(StandardCharsets.UTF_8)
                                , new StringDecoder(StandardCharsets.UTF_8)
                                //自定义处理器
                                , serverHandler);
                    }

                });
        // 绑定端口， 同步等待成功
        ChannelFuture future;
        try {
            log.info("netty服务器在[{}]端口启动监听", port);
            //让netty跑起来的核心代码
            future = bootstrap.bind(port).sync();
            if (future.isSuccess()) {
                nioServerSocketChannel = (NioServerSocketChannel) future.channel();
                log.info("netty服务开启成功" + nioServerSocketChannel);
            } else {
                log.info("netty服务开启失败");
            }
            // 等待服务监听端口关闭,就是由于这里会将线程阻塞，导致无法发送信息，所以我这里开了线程
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //优雅退出，释放线程池资源
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }

    }

}

编写服务处理器

@Component
@ChannelHandler.Sharable
public class ServerHandler extends ChannelInboundHandlerAdapter {

    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    /**
     * 用来保存对应的设备-channel
     */
    private final Map<String, ChannelHandlerContext> channelMap = new ConcurrentHashMap<>();
    /**
     * 用来标记channel当连接断开时要清除channelMap中的记录
     */
    private final Map<ChannelHandlerContext, String> mark = new ConcurrentHashMap<>();

    /**
     * 在与客户端的连接已经建立之后将被调用
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        log.info("new gateway socket from {}", channel.remoteAddress());
    }

    /**
     * 当从客户端接收到一个消息时被调用
     * msg 就是硬件传送过来的数据信息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //这是下面自己写的业务逻辑处理的方法
        String s = msg.toString();
        String message = s.substring(1, s.length() - 1);
        System.out.println("gateway info : " + message);
        String code = message.substring(0, 4);
        boolean containsKey = channelMap.containsKey(code);
        if (!containsKey) {
            channelMap.put(code, ctx);
            mark.put(ctx, code);
        }
        ctx.flush();
    }

    /**
     * 主动发送信息
     *
     * @param code code
     * @param msg  信息
     */
    public void send(String code, String msg) {
        if (channelMap.containsKey(code)) {
            ChannelHandlerContext handlerContext = channelMap.get(code);
            handlerContext.writeAndFlush(msg);
        } else {
            System.out.println("-------设备已经断开连接-------");
        }
    }


    /**
     * 客户端与服务端断开连接时调用
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("netty客户端与服务端连接关闭..." + ctx);
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
                .addListener(ChannelFutureListener.CLOSE);
        boolean containsKey = mark.containsKey(ctx);
        if (containsKey) {
            String code = mark.get(ctx);
            channelMap.remove(code, ctx);
            mark.remove(ctx);
        }
    }

    /**
     * 服务端接收客户端发送过来的数据结束之后调用
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
        log.info("信息接收完毕...");
    }

    /**
     * 在处理过程中引发异常时被调用
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
        System.out.println("异常信息：rn " + cause.getMessage());
    }

}

编写解析器

public class DecoderHandler extends ByteToMessageDecoder {

    private static Map<ChannelHandlerContext, String> msgBufMap = new ConcurrentHashMap<>();

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        byte[] data = new byte[in.readableBytes()];
        in.readBytes(data);
        String msg = new String(data, StandardCharsets.UTF_8);
        if (msg.startsWith("#")) {
            if (msg.endsWith("#")) {
                out.add(msg);
            } else {
                msgBufMap.put(ctx, msg);
            }
        } else if (msg.endsWith("#") && msgBufMap.containsKey(ctx)) {
            msg = msgBufMap.get(ctx) + msg.split("#")[0];
            out.add(msg);
            msgBufMap.remove(ctx);
        }
    }
}

启动Netty服务

在主线程启动后，调用start启动netty服务

public class InitNettyServer implements CommandLineRunner {

    @Autowired
    private NettyServer nettyServer;

    @Autowired
    public void setNettyServer(NettyServer nettyServer) {
        this.nettyServer = nettyServer;
    }

    @Override
    public void run(String... args) throws Exception {
        nettyServer.start();
    }
}

配置文件application.yml

# 应用名称
spring:
  application:
    name: netty

# 应用服务 WEB 访问端口
server:
  port: 8010
netty:
  port: 7002

所需要依赖pom

<properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <netty-all-version>4.1.65.Final</netty-all-version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>${netty-all-version}</version>
        </dependency>

        <!-- 解码and编码器 -->
        <dependency>
            <groupId>org.msgpack</groupId>
            <artifactId>msgpack</artifactId>
            <version>0.6.12</version>
        </dependency>
        <dependency>
            <groupId>org.example</groupId>
            <artifactId>netty</artifactId>
            <version>1.0-SNAPSHOT</version>
            <scope>compile</scope>
        </dependency>

    </dependencies>

编写测试程序

@RestController
public class SendTest {

    @Autowired
    private ServerHandler serverHandler;

    @GetMapping("/send/{code}")
    public String send(@PathVariable(value = "code") String code) {
        serverHandler.send(code,"123456789");
        return "o";
    }
}

这里我采用了api方式调用测试主动发送是否成功

启动程序

@EnableAsync
@SpringBootApplication
public class NettyNioApplication {

    public static void main(String[] args) {
        SpringApplication.run(NettyNioApplication.class, args);
    }
}