SpringBoot 2.7 集成 Netty 4 实现 UDP 通讯

这篇具有很好参考价值的文章主要介绍了SpringBoot 2.7 集成 Netty 4 实现 UDP 通讯。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1 摘要

Netty 作为异步通讯框架,支持多种协议。本文将介绍基于 SpringBoot 2.7 整合 Netty 4 实现 UDP 通讯。

2 核心 Maven 依赖

demo-netty-server/pom.xml
        <!-- Netty -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>${netty.version}</version>
        </dependency>

netty 版本:

<netty.version>4.1.96.Final</netty.version>

3 核心代码

3.1 服务端事务处理器(DemoUdpNettyServerHandler)

demo-netty-server/src/main/java/com/ljq/demo/springboot/netty/server/handler/DemoUdpNettyServerHandler.java
package com.ljq.demo.springboot.netty.server.handler;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


/**
 * @Description: UDP Netty 服务端事务处理器
 * @Author: junqiang.lu
 * @Date: 2023/8/25
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class DemoUdpNettyServerHandler extends SimpleChannelInboundHandler<DatagramPacket> {

    /**
     * 工作线程池
     */
    private final ExecutorService executorService = new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(10000), new DefaultThreadFactory("UDP-netty-work-pool"),
            new ThreadPoolExecutor.CallerRunsPolicy());

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
        ByteBuf byteBuf = packet.content();
        byte[] bytes = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(bytes);
        // 异步处理业务
        executorService.execute(() -> {
            // 读取数据
            log.info("UDP server receive client msg:" + new String(bytes));
            try {
                // 添加休眠,模拟业务处理
                Thread.sleep(5L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("UDP server read message error", cause);
    }
}

代码说明: 这里使用线程池来异步处理事务,提高系统并发性能

3.2 服务端连接类(InitUdpNettyServer)

demo-netty-server/src/main/java/com/ljq/demo/springboot/netty/server/init/InitUdpNettyServer.java
package com.ljq.demo.springboot.netty.server.init;

import com.ljq.demo.springboot.netty.server.handler.DemoUdpNettyServerHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;

/**
 * @Description: 初始化 udt netty 服务
 * @Author: junqiang.lu
 * @Date: 2023/8/25
 */
@Slf4j
@Component
public class InitUdpNettyServer implements ApplicationRunner {

    @Value("${netty.portUdp:9130}")
    private Integer nettyPort;

    @Resource
    private DemoUdpNettyServerHandler udpNettyServerHandler;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        this.start();
    }

    /**
     * 启动服务
     *
     * @throws InterruptedException
     */
    public void start() throws InterruptedException {
        // 连接管理线程池
        EventLoopGroup mainGroup = new NioEventLoopGroup(2);
        EventLoopGroup workGroup = new NioEventLoopGroup(8);
        // 工作线程池
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(mainGroup)
                // 指定 nio 通道,支持 UDP
                .channel(NioDatagramChannel.class)
                // 广播模式
                .option(ChannelOption.SO_BROADCAST, true)
                // 设置读取缓冲区大小为 10M
                .option(ChannelOption.SO_RCVBUF, 1024 * 1024 * 10)
                // 设置发送缓冲区大小为 10M
                .option(ChannelOption.SO_SNDBUF, 1024 * 1024 * 10)
                // 线程池复用缓冲区
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                // 指定 socket 地址和端口
                .localAddress(new InetSocketAddress(nettyPort))
                // 添加通道 handler
                .handler(new ChannelInitializer<NioDatagramChannel>() {
                    @Override
                    protected void initChannel(NioDatagramChannel nioDatagramChannel) throws Exception {
                        nioDatagramChannel.pipeline()
                                // 指定工作线程,提高并发性能
                                .addLast(workGroup,udpNettyServerHandler);
                    }
                });
        // 异步绑定服务器,调用sync()方法阻塞等待直到绑定完成
        bootstrap.bind().sync();
        log.info("---------- [init] UDP netty server start ----------");
    }

}

代码说明:

UDP 协议需要使用 NioDatagramChannel.class 通道

设置缓冲区的大小有利于提高系统吞吐量,线程池复用也利于提升系统处理性能

3.3 客户端事务处理类(DemoUdpNettyClientHandler)

demo-netty-server/src/main/java/com/ljq/demo/springboot/netty/server/handler/DemoUdpNettyClientHandler.java
package com.ljq.demo.springboot.netty.server.handler;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @Description: UDP Netty 客户端事务处理器
 * @Author: junqiang.lu
 * @Date: 2023/8/25
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class DemoUdpNettyClientHandler extends SimpleChannelInboundHandler<DatagramPacket> {


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket) throws Exception {
        // 读取数据
        ByteBuf byteBuf = datagramPacket.content();
        byte[] bytes = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(bytes);
        log.info("receive server msg:" + new String(bytes));
    }
}

3.4 客户端连接类(DemoUdpNettyClient)

demo-netty-server/src/main/java/com/ljq/demo/springboot/netty/server/client/DemoUdpNettyClient.java
package com.ljq.demo.springboot.netty.server.client;

import cn.hutool.core.util.RandomUtil;
import com.ljq.demo.springboot.netty.server.handler.DemoUdpNettyClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
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.DatagramPacket;
import io.netty.channel.socket.nio.NioDatagramChannel;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;

/**
 * @Description: UDP netty 客户端
 * @Author: junqiang.lu
 * @Date: 2023/8/25
 */
@Slf4j
public class DemoUdpNettyClient {

    private final String serverHost;

    private final int serverPort;

    private final int clientPort;

    private final EventLoopGroup mainGroup;

    private final Bootstrap bootstrap;

    private Channel channel;

    public DemoUdpNettyClient(String serverHost, int serverPort, int clientPort) {
        this.serverHost = serverHost;
        this.serverPort = serverPort;
        this.clientPort = clientPort;
        this.mainGroup = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();

    }

    public Channel getChannel() {
        return this.channel;
    }

    /**
     * 创建连接
     */
    public void connect() throws InterruptedException {
        bootstrap.group(mainGroup)
                .channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .localAddress(clientPort)
                .handler(new DemoUdpNettyClientHandler());
        ChannelFuture future = bootstrap.bind().sync();
        this.channel = future.channel();
    }

    /**
     * 发送消息
     *
     * @param message
     */
    public void sendMessage(String message) {
        log.info("客户端待发送消息:{}", message);
        Channel channel = this.getChannel();
        byte[] resBytes = message.getBytes();
        DatagramPacket sendPacket = new DatagramPacket(Unpooled.copiedBuffer(resBytes), new InetSocketAddress(serverHost, serverPort));
        channel.writeAndFlush(sendPacket);
    }

    public void close() throws InterruptedException {
        log.info("关闭客户端");
        mainGroup.shutdownGracefully();
    }




    public static void main(String[] args) throws InterruptedException {
        String serverHost = "127.0.0.1";
        int serverPort = 9130;
        int clientPort = 9131;
        String message = RandomUtil.randomString(1024);
        DemoUdpNettyClient nettyClient = new DemoUdpNettyClient(serverHost, serverPort, clientPort);
        nettyClient.connect();
        for (int i = 0; i < 10000; i++) {
            nettyClient.sendMessage(message + i);
        }
        log.info("--------开始休眠 5 秒------------");
        Thread.sleep(5000L);
        log.info("--------休眠 5 秒结束------------");
        for (int i = 0; i < 5; i++) {
            nettyClient.sendMessage(i + message);
            Thread.sleep(100L);
        }
        Thread.sleep(5000L);
        nettyClient.close();
    }

}

这里包含了测试方法

4 高并发性能配置

  • 1 在服务端事务处理类中使用异步处理消息

  • Netty 服务端设置较高的读写缓存,提高吞吐量;

  • 线程池复用缓冲区

                    // 设置读取缓冲区大小为 10M
                    .option(ChannelOption.SO_RCVBUF, 1024 * 1024 * 10)
                    // 设置发送缓冲区大小为 10M
                    .option(ChannelOption.SO_SNDBUF, 1024 * 1024 * 10)
                    // 线程池复用缓冲区
                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
    
  • Netty 设置主现成以及工作线程,提升消息处理效率

    // 连接管理线程池
            EventLoopGroup mainGroup = new NioEventLoopGroup(2);
            EventLoopGroup workGroup = new NioEventLoopGroup(8);
            // 工作线程池
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(mainGroup)
            
            ... ...
            
            // 添加通道 handler
                    .handler(new ChannelInitializer<NioDatagramChannel>() {
                        @Override
                        protected void initChannel(NioDatagramChannel nioDatagramChannel) throws Exception {
                            nioDatagramChannel.pipeline()
                                    // 指定工作线程,提高并发性能
                                    .addLast(workGroup,udpNettyServerHandler);
                        }
                    });
            
    

5 推荐参考资料

基于Netty实现UDP双向通信

Java入门:UDP协议发送/接收数据实现

读取tcp/udp默认缓冲区大小

Netty之UDP丢包解决

6 Github 源码

Gtihub 源码地址 : https://github.com/Flying9001/springBootDemo/tree/master/demo-netty-server

个人公众号:404Code,分享半个互联网人的技术与思考,感兴趣的可以关注.
文章来源地址https://www.toymoban.com/news/detail-771877.html

到了这里,关于SpringBoot 2.7 集成 Netty 4 实现 UDP 通讯的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Netty系列(一):Springboot整合Netty,自定义协议实现

    Netty是由JBOSS提供的一个java开源框架,现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具,用以快速开发高性能、高可靠性的网络服务器和客户端程序。 也就是说,Netty 是一个基于NIO的客户、服务器端的编程框架,使用Netty 可以确保你快速和简单

    2023年04月25日
    浏览(53)
  • Springboot整合Netty,自定义协议实现

    新建springboot项目,并在项目以来中导入netty包,用fastjson包处理jsonStr。 创建netty相关配置信息文件 yml配置文件—— application.yml netty配置实体类—— NettyProperties 与yml配置文件绑定 通过 @ConfigurationProperties(prefix = \\\"netty\\\") 注解读取配置文件中的netty配置,通过反射注入值,需要在

    2024年02月06日
    浏览(38)
  • SpringBoot+Netty+Websocket实现消息推送

    这样一个需求:把设备异常的状态每10秒推送到页面并且以弹窗弹出来,这个时候用Websocket最为合适,今天主要是后端代码展示。 添加依赖 定义netty端口号 netty服务器 Netty配置 管理全局Channel以及用户对应的channel(推送消息) 管道配置 自定义CustomChannelHandler 推送消息接口及

    2024年02月04日
    浏览(47)
  • netty学习(3):SpringBoot整合netty实现多个客户端与服务器通信

    创建一个SpringBoot工程,然后创建三个子模块 整体工程目录:一个server服务(netty服务器),两个client服务(netty客户端) pom文件引入netty依赖,springboot依赖 NettySpringBootApplication NettyServiceHandler SocketInitializer NettyServer NettyStartListener application.yml Client1 NettyClientHandler SocketInitializ

    2024年02月11日
    浏览(56)
  • SpringBoot整合Netty+Websocket实现消息推送

           Netty是一个高性能、异步事件驱动的网络应用框架,用于快速开发可维护的高性能协议服务器和客户端。以下是Netty的主要优势: 高性能 :Netty基于NIO(非阻塞IO)模型,采用事件驱动的设计,具有高性能的特点。它通过零拷贝技术、内存池化技术等手段,进一步提高

    2024年01月20日
    浏览(43)
  • Springboot中使用netty 实现 WebSocket 服务

    依赖 创建启动类 创建WebSocket 服务 WsServerInitialzer 初始化 创建信息ChatHandler 处理类

    2024年02月14日
    浏览(37)
  • Springboot整合Netty实现RPC服务器

    try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(boss, worker) .childHandler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new IdleStateHandler(0, 0, 60)); pipeline.addLast(new JsonDecoder()); pipeline.addLast(new JsonEnco

    2024年04月09日
    浏览(39)
  • 基于Springboot用Netty实现WebSocket及用户身份校验

    说在前头,文本主要参考: SpringBoot+WebSocket+Netty实现消息推送 Netty-11-channelHandler的生命周期 springboot整合netty指北 首先 需要了解下channel建立的生命周期 ChannelHandler的顺序如下: 注意本次实现的重点是:在建立websocket时从请求标头header或者第一次消息对话时获取用户信息(如jw

    2024年02月04日
    浏览(41)
  • SpringBoot+Netty+Vue+Websocket实现在线推送/聊天系统

    ok,那么今天的话也是带来这个非常常用的一个技术,那就是咱们完成nutty的一个应用,今天的话,我会介绍地很详细,这样的话,拿到这个博文的代码就基本上可以按照自己的想法去构建自己的一个在线应用了。比如聊天,在线消息推送之类的。其实一开始我原来的想法做在

    2024年02月03日
    浏览(39)
  • 基于Springboot+WebSocket+Netty实现在线聊天、群聊系统

    此文主要实现在好友添加、建群、聊天对话、群聊功能,使用Java作为后端语言进行支持,界面友好,开发简单。 2.1、下载安装IntelliJ IDEA(后端语言开发工具),Mysql数据库,微信Web开发者工具。 1.创建maven project 先创建一个名为SpringBootDemo的项目,选择【New Project】 然后在弹出

    2024年02月14日
    浏览(41)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包