springboot+netty化身Udp服务端，go化身客户端模拟设备实现指令联动

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.69.Final</version> <!-- 替换为最新版本 -->
</dependency>

package com.todoitbo.baseSpringbootDasmart.netty.server;

import com.todoitbo.baseSpringbootDasmart.netty.handler.UdpHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;

/**
 * @author todoitbo
 * @date 2023/11/29
 */
@Slf4j
public class NettyUdpServer {

    private final int nettyPort;

    public static Channel channel;

    public NettyUdpServer(int port) {
        this.nettyPort = port;
    }

    /**
     * 启动服务
     *
     * @throws InterruptedException
     */
    public void start() throws InterruptedException {
        // 连接管理线程池
        EventLoopGroup mainGroup = new NioEventLoopGroup(2);
        EventLoopGroup workGroup = new NioEventLoopGroup(8);

        try {
            // 工作线程池
            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,new UdpHandler());
                        }
                    });

            // 异步绑定服务器，调用sync()方法阻塞等待直到绑定完成
            ChannelFuture sync = bootstrap.bind().sync();
            channel = sync.channel();
            log.info("---------- [init] UDP netty server start ----------");

            // 阻塞等待服务器关闭
            channel.closeFuture().sync();
        } finally {
            // 释放资源
            mainGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

package com.todoitbo.baseSpringbootDasmart.netty.handler;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import static com.todoitbo.baseSpringbootDasmart.controller.SysUploadController.socketAddressMap;

/**
 * @author todoitbo
 * @date 2023/11/29
 */
public class UdpHandler extends SimpleChannelInboundHandler<DatagramPacket> {

    // 使用 CompletableFuture 用于异步获取客户端的响应
    public static CompletableFuture<String> responseFuture = new CompletableFuture<>();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
        // 从DatagramPacket中获取数据和发送者信息
        byte[] data;
        int len = packet.content().readableBytes();

        if (packet.content().hasArray()) {
            data = packet.content().array();
        } else {
            data = new byte[len];
            packet.content().getBytes(packet.content().readerIndex(), data);
        }
        String senderAddress = packet.sender().getAddress().getHostAddress();
        int senderPort = packet.sender().getPort();

        // 处理接收到的数据
        String message = new String(data);
        System.out.println("Received message from " + senderAddress + ":" + senderPort + " - " + message);

        if (message.contains("test")) {
            responseFuture.complete(message);
        }

        // 构建响应消息
        String response = "Hello, client!";
        byte[] responseData = response.getBytes();

        // 创建响应的DatagramPacket并发送给发送者
        InetSocketAddress senderSocketAddress = new InetSocketAddress(senderAddress, senderPort);
        socketAddressMap.put("test", senderSocketAddress);
        DatagramPacket responsePacket = new DatagramPacket(Unpooled.copiedBuffer(responseData), senderSocketAddress);
        ctx.writeAndFlush(responsePacket);
    }

    /*@Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 处理异常情况
        cause.printStackTrace();
        ctx.close();
    }*/
    // 在接口调用后等待客户端响应的方法
    public static String waitForClientResponse() {
        try {
            // 使用 CompletableFuture 的 get 方法来阻塞等待客户端的响应
            String s = responseFuture.get(500, TimeUnit.MILLISECONDS);
            responseFuture = new CompletableFuture<>();
            return s; // 等待时间为 1 秒
        } catch (Exception e) {
            // 发生超时或其他异常，可以根据实际情况处理
            return "456"; // 超时返回默认值 "456"
        }
    }
}

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class BaseSpringbootDasmartApplication {
   /* static {
        AspectLogEnhance.enhance();}//进行日志增强，自动判断日志框架*/
    public static void main(String[] args) {
        // System.setProperty("log4j2.isThreadContextMapInheritable", Boolean.TRUE.toString());
        SpringApplication.run(BaseSpringbootDasmartApplication.class, args);
        try {
            // new NettyWebsocketServer(13025).run();
            new NettyUdpServer(13026).start();
        } catch (Exception e) {
            throw new BusinessException("-----启动失败-----", e.getMessage()).setCause(e).setLog();
        }
    }
}

@GetMapping("/login/{message}")
public String login(@PathVariable String message) throws NacosException, InterruptedException {

    byte[] responseData = message.getBytes();

    // 创建响应的DatagramPacket并发送给发送者
    DatagramPacket responsePacket = new DatagramPacket(Unpooled.copiedBuffer(responseData), socketAddressMap.get("test"));
    NettyUdpServer.channel.writeAndFlush(responsePacket);
    // 客户端是否响应，响应返回传入值，否则返回456，响应时间不超过0.5s，如果10.5s还未响应，则返回456
    return UdpHandler.waitForClientResponse();
}

// @Author todoitbo 2023/11/29 14:26:00
package utils

import (
    "context"
    "fmt"
    "net"
    "strings"
    "sync"
)

// UDPClient 是一个简单的 UDP 客户端
type UDPClient struct {
    conn *net.UDPConn
    mu   sync.Mutex
}

// NewUDPClient 创建一个新的 UDP 客户端
func NewUDPClient(serverAddr string) (*UDPClient, error) {
    client := &UDPClient{}

    addr, err := net.ResolveUDPAddr("udp", serverAddr)
    if err != nil {
       return nil, err
    }

    conn, err := net.DialUDP("udp", nil, addr)
    if err != nil {
       return nil, err
    }

    client.conn = conn
    message := []byte("你好")
    _, err = conn.Write(message)
    return client, nil
}

// Close 关闭 UDP 客户端连接
func (c *UDPClient) Close() {
    c.mu.Lock()
    defer c.mu.Unlock()
    if c.conn != nil {
       c.conn.Close()
    }
}

// ListenForMessages 启动 Goroutine 监听服务端的实时消息
func (c *UDPClient) ListenForMessages(ctx context.Context, wg *sync.WaitGroup) {
    defer wg.Done() // 在 Goroutine 结束时通知 WaitGroup

    buffer := make([]byte, 1024)
    for {
       select {
       case <-ctx.Done():
          // 收到关闭信号，结束 Goroutine
          return
       default:
          c.mu.Lock()
          conn := c.conn
          c.mu.Unlock()

          if conn == nil {
             // 客户端连接已关闭
             return
          }

          n, _, err := conn.ReadFromUDP(buffer)
          if err != nil {
             fmt.Println("Error reading from server:", err)
             return
          }

          // 处理收到的消息，可以根据实际需求进行逻辑处理
          message := string(buffer[:n])
          if strings.Contains(message, "test") {
             c.SendMessage(message)
          }

          // hexString := hex.EncodeToString(message)
          // 将 3600 转换为字符串
          /*expectedValue := "3600"
          if hexString == expectedValue {
             c.SendMessage("test")
          }else {
             c.SendMessage(message)
          }*/
          fmt.Println("Received message from server: ", message)
       }
    }
}

// SendMessage 向服务端发送消息
func (c *UDPClient) SendMessage(message string) error {
    c.mu.Lock()
    defer c.mu.Unlock()

    if c.conn == nil {
       return fmt.Errorf("client connection is closed")
    }

    _, err := c.conn.Write([]byte(message))
    return err
}

func InitUDPClient(ctx context.Context, wg *sync.WaitGroup, serverAddr string) (*UDPClient, error) {
    client, err := NewUDPClient(serverAddr)
    if err != nil {
       return nil, err
    }

    // 启动 Goroutine 监听服务端的实时消息
    wg.Add(1)
    go client.ListenForMessages(ctx, wg)

    return client, nil
}

func init() {
    InitUDPClient(context.Background(), &sync.WaitGroup{}, "127.0.0.1:13026")
}