SpringBoot+UDPClient+定時任務通信實戰

在springboot配置文件中添加需要連接的服務器地址及端口號

#udp連接的服務器地址與端口
host=10.1.7.113
port=1215

UDP客戶端編寫

package com.example.udp.client;

import io.netty.bootstrap.Bootstrap;
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.nio.NioDatagramChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * udpclient
 *
 * @author LH
 * @date 2021/12/20 9:53
 */

@Component
public class NettyUdpClient {

    private static final Logger log = LoggerFactory.getLogger(NettyUdpClient.class);

    @Value("${host}")
    private String HOST;
    @Value("${port}")
    private int PORT;

    @Autowired
    ClientChannelInitializer clientChannelInitializer;

    //與服務端建立連接后得到的通道對象
    private Channel channel;

    /**
     * 初始化 `Bootstrap` 客戶端引導程序
     *
     * @return
     */
    private final Bootstrap getBootstrap() {
        Bootstrap b = new Bootstrap();
        EventLoopGroup group = new NioEventLoopGroup();
        b.group(group)
                .channel(NioDatagramChannel.class)//數據包通道，udp通道類型
                .handler(clientChannelInitializer)//通道處理者
                .option(ChannelOption.SO_BROADCAST, true);//開啟廣播
        return b;
    }

    /**
     * 建立連接，獲取連接通道對象
     *
     * @return
     */
    public void connect() {
        ChannelFuture channelFuture = getBootstrap().connect(HOST, PORT).syncUninterruptibly();
        if (channelFuture != null && channelFuture.isSuccess()) {
            log.warn("udp client connect host = {}, port = {} success", HOST, PORT);
            channel = channelFuture.channel();
        } else {
            log.error("udp client connect host = {}, port = {} failed!", HOST, PORT);
        }
    }

}

添加自定義通道處理器

package com.example.udp.client;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.nio.NioDatagramChannel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 通道初始化，主要用於設置各種Handler
 *
 * @author LH
 * @date 2021/12/20 9:55
 */
@Component
public class ClientChannelInitializer extends ChannelInitializer<NioDatagramChannel> {

    @Autowired
    ClientChannelHandler clientChannelHandler;

    @Override
    protected void initChannel(NioDatagramChannel nioDatagramChannel) throws Exception {

        ChannelPipeline pipeline = nioDatagramChannel.pipeline();
        //自定義Handler
        pipeline.addLast("clientChannelHandler", clientChannelHandler);
    }

}

自定義數據處理handler編寫

package com.example.udp.client;

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.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * 數據處理handler
 *
 * @author LH
 * @date 2021/12/20 9:56
 */

@Component
@ChannelHandler.Sharable
@Slf4j
public class ClientChannelHandler extends SimpleChannelInboundHandler<DatagramPacket> {

    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private ScheduledFuture<?> heartBeat;

    /**
     * 通道建立成功后
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 當channel就緒后
        log.info("[UDP] client channel is ready!");
        // 定時任務編寫
        this.heartBeat = this.scheduler.scheduleWithFixedDelay(new ClientMain(ctx), 0, 2, TimeUnit.SECONDS);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket) throws Exception {
        System.out.println("[UDP] client 收到的消息：" + datagramPacket.content().toString(CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

}

自定義類實現Runnable接口，開一個線程進行邏輯處理

package com.bwss.collectoroutlineserver.client;

import com.alibaba.fastjson.JSON;
import com.bwss.common.utils.ChannelMap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.CharsetUtil;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * 數據業務邏輯處理
 *
 * @author LH
 * @date 2021/12/20 15:26
 */
public class ClientMain implements Runnable{

    private final ChannelHandlerContext ctx;

    public ClientMain(ChannelHandlerContext ctx) {
        this.ctx = ctx;
    }

    @Override
    public void run() {
        
       // 業務邏輯編寫


        // 將所有存活的設備id傳送給業務端
        ctx.writeAndFlush(Unpooled.copiedBuffer(JSON.toJSONString(需要進行傳輸的數據), CharsetUtil.UTF_8));
    }
}

主啟動類編寫，實現CommandLineRunner接口，在run方法中跑udp客戶端

public class CollectorOutlineServerApplication implements CommandLineRunner
{

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

    @Autowired
    NettyUdpClient nettyUdpClient;

    @Override
    public void run(String... args) throws Exception {
        // 啟動udp客戶端
        nettyUdpClient.connect();
    }
}

注意：

上述方法是采用Spring注入的方式進行的，你也可以使用 new 對象的方式進行，效果都是一樣的。

上面是udp通信客戶端的編寫，服務器端需要處理繁雜的業務邏輯，因此可以自定義實現，下面給出簡單的參考示例。

udp服務端簡單示例

package com.example.udp.server;

import io.netty.bootstrap.Bootstrap;
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.nio.NioDatagramChannel;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;

/**
 * udp服務端
 *
 * @author LH
 * @date 2021/12/20 9:45
 */

@Component
public class NettyUdpServer {

    private static final Logger log = LoggerFactory.getLogger(NettyUdpServer.class);

    private static final EventLoopGroup group = new NioEventLoopGroup(1);

    @Autowired
    ServerChannelInitializer serverChannelInitializer;

    @Value("${port}")
    private int port;

    //監聽端口的通道，即server的處理通道
    private Channel channel;

    /**
     * 開啟udp server服務
     *
     * @return
     */
    public ChannelFuture start() {
        //啟動類
        Bootstrap serverBootstrap = new Bootstrap();
        serverBootstrap.group(group)//組配置，初始化ServerBootstrap的線程組
                .channel(NioDatagramChannel.class)//數據包通道，udp通道類型
                .option(ChannelOption.SO_BROADCAST, true)//支持廣播
                .handler(serverChannelInitializer);//通道處理者
        //Future：異步任務的生命周期，可用來獲取任務結果
        ChannelFuture channelFuture1 = serverBootstrap.bind(port).syncUninterruptibly();//綁定端口，開啟監聽,同步等待
        if (channelFuture1 != null && channelFuture1.isSuccess()) {
            log.info("[UDP] server start success, port = {}", port);
            channel = channelFuture1.channel();//獲取通道
        } else {
            log.error("udp server start failed!!");
            channelFuture1.cause().printStackTrace();
        }
        return channelFuture1;
    }

    /**
     * 停止udp server服務
     * 銷毀前的攔截
     */
    @PreDestroy
    public void destroy() {
        try {
            if (channel != null) {
                ChannelFuture await = channel.close().await();
                if (!await.isSuccess()) {
                    log.error("udp channel close fail, {}", await.cause());
                }
            }
            Future<?> future1 = group.shutdownGracefully().await();
            if (!future1.isSuccess()) {
                log.error("udp group shutdown fail, {}", future1.cause());
            }
            log.info("udp shutdown success");
        } catch (InterruptedException e) {
            log.info("udp shutdown fail");
            e.printStackTrace();
        }
    }
}

package com.example.udp.server;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.nio.NioDatagramChannel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 自定義ChannelInitializer,通道初始化，主要用於設置各種Handler
 *
 * @author LH
 * @date 2021/12/20 9:49
 */
@Component
public class ServerChannelInitializer extends ChannelInitializer<NioDatagramChannel> {

    @Autowired
    ServerChannelInboundHandler serverChannelHandler;

    @Override
    protected void initChannel(NioDatagramChannel nioDatagramChannel) throws Exception {

        ChannelPipeline pipeline = nioDatagramChannel.pipeline();

        pipeline.addLast("serverChannelHandler", serverChannelHandler);
    }

}

package com.example.udp.server;

import io.netty.buffer.Unpooled;
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.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * 通道數據輸入的處理
 *
 * @author LH
 * @date 2021/12/20 9:51
 */
@Component
@ChannelHandler.Sharable
@Slf4j
public class ServerChannelInboundHandler extends SimpleChannelInboundHandler<DatagramPacket> {

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket) throws Exception {
        System.out.println("[UDP] server 收到的消息：" + datagramPacket.content().toString(CharsetUtil.UTF_8));
        String response = "{" + datagramPacket.content().toString(CharsetUtil.UTF_8) + "}的響應，我是服務端啊！！！";
        DatagramPacket datagramPacket1 = new DatagramPacket(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8), datagramPacket.sender());
        channelHandlerContext.channel().writeAndFlush(datagramPacket1);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

}

server.port=8090