[Spring Boot]Netty-UDP客户端_netty udp客户端

简述

最近在一些场景中需要使用UDP客户端进行，所以开始集成新的东西。本文集成了一个基于netty的SpringBoot的简单的应用场景。

Netty-UDP集成

pom引入

        <!-- netty -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.84.Final</version>
        </dependency>

1
2
3
4
5
6
7

如果项目本身使用redis,则redis默认会集成,我的项目中默认有netty-4.1.84.Final

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
1
2
3
4

Client

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;


/**
 * @author fir
 */
@Slf4j
@Component
public class LogPushUdpClient implements ApplicationRunner {

    private final Bootstrap bootstrap;
    public final NioEventLoopGroup workerGroup;
    public static Channel channel;

    @Override
    public void run(ApplicationArguments args) {
        start();
    }

    public void start() {
        try {
            log.info("UDP客户端--启动");
            channel = bootstrap
            .bind(1234)
            .sync()
            .channel();
            channel.closeFuture().await(1000);
        } catch (InterruptedException e) {
            log.info("UDP客户端启动失败");
        }
    }

    private LogPushUdpClient() {
        bootstrap = new Bootstrap();
        workerGroup = new NioEventLoopGroup();
        bootstrap.group(workerGroup)
        .channel(NioDatagramChannel.class)
        .option(ChannelOption.SO_BROADCAST, true)
        .handler(new ChannelInitializer<NioDatagramChannel>() {

            @Override
            protected void initChannel(NioDatagramChannel ch) {
                // 获取了通道的管道，管道是一个处理网络事件和操作的处理器链。
                ChannelPipeline pipeline = ch.pipeline();
                // 向管道添加了自定义的 LogPushUdpClientHandler 处理器，它将处理网络 I/O 事件和数据
                pipeline.addLast(new LogPushUdpClientHandler());
            }
        });
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

Handler

package com.fir.home.handler.udp;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.concurrent.GenericFutureListener;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;

import lombok.extern.slf4j.Slf4j;



/**
 * @author fir
 */
@Slf4j
public class LogPushUdpClientHandler extends SimpleChannelInboundHandler<DatagramPacket> {


    public static HashMap<Long, Boolean> treedMap = new HashMap<>();

    /**
     * 客户端初次连接时执行的方法
     *
     * @param ctx 通道处理上下文
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        log.info("客户端通道已就绪!");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) {
        final ByteBuf buf = packet.content();
        int readableBytes = buf.readableBytes();
        byte[] content = new byte[readableBytes];
        buf.readBytes(content);
        String serverMessage = new String(content);
        log.info("[UDP客户端]接受消息: " + serverMessage);
    }


    /**
     * 向服务器发送消息
     *
     * @param msg               按规则拼接的消息串
     * @param inetSocketAddress 目标服务器地址
     */
    public static void sendMessage(final String msg, final InetSocketAddress inetSocketAddress) {
        log.info("[UDP客户端]发送消息消息: " + msg);
        if (msg == null) {
            throw new NullPointerException("[UDP客户端]发送数据为空");
        }
        DatagramPacket datagramPacket = datagramPacket(msg, inetSocketAddress);
        senderInternal(datagramPacket);
    }


    /**
     * 组装数据包
     *
     * @param msg               消息串
     * @param inetSocketAddress 服务器地址
     * @return DatagramPacket
     */
    public static DatagramPacket datagramPacket(String msg, InetSocketAddress inetSocketAddress) {
        DatagramPacket datagramPacket;
        ByteBuf dataBuf = Unpooled.copiedBuffer(msg, StandardCharsets.UTF_8);
        datagramPacket = new DatagramPacket(dataBuf, inetSocketAddress);
        return datagramPacket;
    }


    /**
     * 发送数据包服务器无返回结果
     *
     * @param datagramPacket 数据报文包
     */
    private static void senderInternal(final DatagramPacket datagramPacket) {
        if (LogPushUdpClient.channel != null) {
            LogPushUdpClient.channel.writeAndFlush(datagramPacket).addListener((GenericFutureListener<ChannelFuture>) future -> {
                boolean success = future.isSuccess();
                if (log.isInfoEnabled()) {
                    log.info("[UDP客户端]发送结果 : " + success);
                }
            });
        } else {
            throw new NullPointerException("UPD上下文通道为空");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

调用

        InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 50000);
        LogPushUdpClientHandler.sendMessage("hello", inetSocketAddress);
1
2

消息发送与接收

在线UDP服务

服务端可以使用在线UDP服务
http://udp.xnkiot.com/

系统调用

调用后，可以看到，数据发送成功，并且接受成功。