物联网之使用Vertx实现UDP最佳实践【响应式】

小伙伴们，你们好呀，我是老寇，跟我一起学习使用Vertx实现UDP-Server
实现UDP【响应式】

Vertx-Core地址
注意

UDP是无连接的传输，这意味着您与远程客户端没有建立持续的连接。
所以，您发送和接收的数据包都要包含有远程的地址。
除此之外，UDP不像TCP的使用那样安全， 这也就意味着不能保证发送的数据包一定会被对应的接收端（Endpoint）接收。【传输数据时不建立连接，因此可能导致数据包丢失】
UDP最适合一些允许丢弃数据包的应用（如监视应用程序，视频直播）。
实现过程

查看源码
代码比较简单，懒得讲解啦

代码比较简单，懒得讲解啦

代码比较简单，懒得讲解啦

服务端

引入依赖

1. <dependency>
2.   <groupId>io.vertx</groupId>
3.   vertx-core</artifactId>
4.   <version>5.0.0</version>
5. </dependency>

复制代码

UdpServerProperties

1. /**
2. * @author laokou
3. */
4. @Data
5. @Component
6. @ConfigurationProperties(prefix = "spring.udp-server")
7. public class UdpServerProperties {
9.     private String host = "0.0.0.0";
11.     private Set<Integer> ports = new HashSet<>(0);
13.     private boolean broadcast = false;
15.     private boolean loopbackModeDisabled = true;
17.     private String multicastNetworkInterface = null;
19.     private boolean ipV6 = false;
21. }

复制代码

VertxUdpServer

1. /**
2. * @author laokou
3. */
4. @Slf4j
5. public final class VertxUdpServer extends AbstractVerticle {
7.     private volatile Flux<DatagramSocket> datagramSocket;
9.     private final UdpServerProperties udpServerProperties;
11.     private boolean isClosed = false;
13.     VertxUdpServer(Vertx vertx, UdpServerProperties udpServerProperties) {
14.        this.udpServerProperties = udpServerProperties;
15.        this.vertx = vertx;
16.     }
18.     @Override
19.     public synchronized void start() {
20.        datagramSocket = Flux.fromIterable(udpServerProperties.getPorts()).map(port -> {
21.           DatagramSocket datagramSocket = vertx.createDatagramSocket(getDatagramSocketOption())
22.              .handler(packet -> log.info("【Vertx-UDP-Server】 => 收到数据包：{}", packet.data()));
23.           datagramSocket.listen(port, udpServerProperties.getHost()).onComplete(result -> {
24.              if (isClosed) {
25.                 return;
26.              }
27.              if (result.succeeded()) {
28.                 log.info("【Vertx-UDP-Server】 => UDP服务启动成功，端口：{}", port);
29.              }
30.              else {
31.                 Throwable ex = result.cause();
32.                 log.error("【Vertx-UDP-Server】 => UDP服务启动失败，错误信息：{}", ex.getMessage(), ex);
33.              }
34.           });
35.           return datagramSocket;
36.        });
37.        datagramSocket.subscribeOn(Schedulers.boundedElastic()).subscribe();
38.     }
40.     @Override
41.     public synchronized void stop() {
42.        isClosed = true;
43.        datagramSocket.doOnNext(socket -> socket.close().onComplete(result -> {
44.           if (result.succeeded()) {
45.              log.info("【Vertx-UDP-Server】 => UDP服务停止成功");
46.           }
47.           else {
48.              Throwable ex = result.cause();
49.              log.error("【Vertx-UDP-Server】 => UDP服务停止失败，错误信息：{}", ex.getMessage(), ex);
50.           }
51.        })).subscribeOn(Schedulers.boundedElastic()).subscribe();
52.     }
54.     public void deploy() {
55.        // 部署服务
56.        vertx.deployVerticle(this);
57.        // 停止服务
58.        Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
59.     }
61.     private DatagramSocketOptions getDatagramSocketOption() {
62.        DatagramSocketOptions datagramSocketOptions = new DatagramSocketOptions();
63.        datagramSocketOptions.setBroadcast(udpServerProperties.isBroadcast());
64.        datagramSocketOptions.setLoopbackModeDisabled(udpServerProperties.isLoopbackModeDisabled());
65.        datagramSocketOptions.setMulticastNetworkInterface(udpServerProperties.getMulticastNetworkInterface());
66.        datagramSocketOptions.setIpV6(udpServerProperties.isIpV6());
67.        return datagramSocketOptions;
68.     }
70. }

复制代码

VertxUdpServerManager

1. /**
2. * @author laokou
3. */
4. public final class VertxUdpServerManager {
6.     private VertxUdpServerManager() {
7.     }
9.     public static void deploy(final Vertx vertx, final UdpServerProperties properties) {
10.        new VertxUdpServer(vertx, properties).deploy();
11.     }
13. }

复制代码

客户端【测试】

1. /**
2. * @author laokou
3. */
4. @Slf4j
5. @SpringBootTest
6. @RequiredArgsConstructor
7. @TestConstructor(autowireMode = TestConstructor.AutowireMode.ALL)
8. class UdpTest {
10.     private final Vertx vertx;
12.     @Test
13.     void test() throws InterruptedException {
14.        for (int i = 4880; i < 5000; i++) {
15.           DatagramSocket datagramSocket = vertx.createDatagramSocket();
16.           int finalI = i;
17.           datagramSocket.send("Hello Vertx", i, "127.0.0.1").onComplete(result -> {
18.              if (result.succeeded()) {
19.                 log.info("【Vertx-UDP-Client】 => 发送成功，端口：{}", finalI);
20.              }
21.              else {
22.                 Throwable ex = result.cause();
23.                 log.error("【Vertx-UDP-Client】 => 发送失败，端口：{}，错误信息：{}", finalI, ex.getMessage(), ex);
24.              }
25.           });
26.           Thread.sleep(2000);
27.           Assertions.assertDoesNotThrow(datagramSocket::close);
28.        }
29.     }
31. }

复制代码

这可以满足基本的协议开发，自行修改即可！！！
我是老寇，我们下次再见啦！

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！