JAVA通过netty实现接收多UDP上报数据再WebSocket推送至前端

2021-09-19

Gradle引入依赖

// netty
implementation("io.netty:netty-all")
// websocket
implementation 'org.springframework.boot:spring-boot-starter-websocket'

在config包下新建WebSocketConfig

import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration;

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig  implements WebSocketMessageBrokerConfigurer {

    @Override
    public void registerStompEndpoints(StompEndpointRegistry stompEndpointRegistry){
        // SockJS + Stomp
        stompEndpointRegistry
                .addEndpoint("/websocket")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry){
        // 服务端向客户端发送消息的前缀
        registry.enableSimpleBroker("/topic");
        registry.setUserDestinationPrefix("/user");
        registry.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registry){
        // 设置缓冲区大小及超时，针对数据长度超过默认值的应用场景
        registry.setMessageSizeLimit(10 * 1024 * 1024);
        registry.setSendBufferSizeLimit(100 * 1024 * 1024);     // 100MB
        registry.setSendTimeLimit(200000);
    }

    @Override
    public void configureClientOutboundChannel(ChannelRegistration registration){
        // 设置默认发送器线程个数为1，保证推送数据的有序性
        // 多线程协作不能保证推送数据的有序性
        registration.taskExecutor().corePoolSize(1).maxPoolSize(1);
    }
}

新建netty包

在controller同级目录新建包netty及其子包

1
2
3

netty/channel
netty/config
netty/thread

/netty/channel/UdpServerConfig

import lombok.Data;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;

@Component
@RefreshScope
@Data
public class UdpServerConfig {

    /**
     * udp服务器端口号：如果配置文件中未设置，默认为18899端口
     */
    private Integer port = 18899;

    /**
     * netty监听线程数，默认为0
     */
    private Integer eventLoopGroupNum = 0;


    /**
     * @Description: udp编码格式
     */
    private String encode = "utf-8";

    /**
     * @Description: udp解码格式
     */
    private String decode = "utf-8";

}

/netty/channel/UdpServer

import com.network.analysis.system.netty.channel.UdpServerChannelInboundHandler;
import com.network.analysis.system.netty.config.UdpServerConfig;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.FixedRecvByteBufAllocator;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class UdpServer implements CommandLineRunner {

    @Resource
    UdpServerConfig udpServerConfig;

    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    @Async
    @Override
    public void run(String... args) throws Exception {
        // 启动UDP服务端
        // 加载完Bean后执行
         start();
    }

    /**
     * udp服务端启动
     */
    private void start() {
        // 服务器连接监听线程组，专门接受accept 新的客户端client连接
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup(udpServerConfig.getEventLoopGroupNum());
        try {
            // netty 的bootstrap启动类
            Bootstrap bootstrap = new Bootstrap();
            // 设置连接监听线程组
            bootstrap.group(eventLoopGroup);
            // 设置连接协议为UDP协议
            bootstrap.channel(NioDatagramChannel.class);
            // 设置通道参数为广播模式
            bootstrap.option(ChannelOption.SO_BROADCAST, true);
            // 设置UDP缓冲区大小
            // UDP包数据长度超过默认值，需另外设置上限
            bootstrap.option(ChannelOption.SO_RCVBUF, 100 * 1024);
            // 设置DatagramPacket msg大小
            // DatagramPacket的长度超过默认值，需另外设置上限
            bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(100 * 1024));
            // 设置消息处理管道
            UdpServerChannelInboundHandler handler = new UdpServerChannelInboundHandler(udpServerConfig, messagingTemplate);
            bootstrap.handler(handler);
            // 设置udp服务端端口启动
            Integer port = udpServerConfig.getPort();
            ChannelFuture f = bootstrap.bind(port).sync();
            System.out.println("netty udp start! port: " + port);
            // 监听连接关闭
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("netty udp close!");
            eventLoopGroup.shutdownGracefully();
        }
    }
}

/netty/channel/UdpServerChannelInboundHandler

import com.network.analysis.system.netty.config.UdpServerConfig;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import org.springframework.messaging.simp.SimpMessagingTemplate;

import java.net.InetSocketAddress;

@EqualsAndHashCode(callSuper = true)
@Data
@AllArgsConstructor
@NoArgsConstructor
public class UdpServerChannelInboundHandler extends SimpleChannelInboundHandler<DatagramPacket> {

    /**
     * UDP参数
     */
    private UdpServerConfig udpServerConfig;

    /**
     * WebSocket消息模板
     */
    private SimpMessagingTemplate messagingTemplate;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
        // 获取UDP连接的地址
        InetSocketAddress socketAddress = msg.sender();
        // 当前连接的host
        String host = socketAddress.getAddress().getHostAddress();
        // 当前连接的端口
        int port = socketAddress.getPort();

        // 发送过来的内容
        ByteBuf content = msg.content();
        byte[] req = new byte[content.readableBytes()];
        content.readBytes(req);
        String string = new String(req,"utf-8");

        // 根据UDP的IP、PORT转发数据到相应的前端订阅接口
        if(port == xxxx && "x.x.x.x".equals(host)) {
            dataTranspond("/topic/xxx", "/xx", string);
        }
        if(port == xxxx && "x.x.x.x".equals(host)) {
            dataTranspond("/topic/xxxx", "/xxx", string);
        }

        // 响应发送方成功接收
        ctx.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer("success", Charset.forName(udpServerConfig.getEncode())), msg.sender()));
    }

    // 数据转发前端订阅接口
    public void dataTranspond(String sendPath, String sendIdentify, String rec){
        // webSock推送数据到前端
        messagingTemplate.convertAndSend(sendPath + sendIdentify, rec);
    }
}