当前位置: 首页 > news >正文

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

小伙伴们,你们好呀,我是老寇,跟我一起学习使用Vertx实现UDP-Server

实现UDP【响应式】

Vertx-Core地址

注意

UDP是无连接的传输,这意味着您与远程客户端没有建立持续的连接。

所以,您发送和接收的数据包都要包含有远程的地址。

除此之外,UDP不像TCP的使用那样安全, 这也就意味着不能保证发送的数据包一定会被对应的接收端(Endpoint)接收。【传输数据时不建立连接,因此可能导致数据包丢失

UDP最适合一些允许丢弃数据包的应用(如监视应用程序,视频直播)。

实现过程

查看源码

代码比较简单,懒得讲解啦
代码比较简单,懒得讲解啦
代码比较简单,懒得讲解啦
服务端
引入依赖
<dependency><groupId>io.vertx</groupId><artifactId>vertx-core</artifactId><version>5.0.0</version>
</dependency>

UdpServerProperties

/*** @author laokou*/
@Data
@Component
@ConfigurationProperties(prefix = "spring.udp-server")
public class UdpServerProperties {private String host = "0.0.0.0";private Set<Integer> ports = new HashSet<>(0);private boolean broadcast = false;private boolean loopbackModeDisabled = true;private String multicastNetworkInterface = null;private boolean ipV6 = false;}

VertxUdpServer

/*** @author laokou*/
@Slf4j
public final class VertxUdpServer extends AbstractVerticle {private volatile Flux<DatagramSocket> datagramSocket;private final UdpServerProperties udpServerProperties;private boolean isClosed = false;VertxUdpServer(Vertx vertx, UdpServerProperties udpServerProperties) {this.udpServerProperties = udpServerProperties;this.vertx = vertx;}@Overridepublic synchronized void start() {datagramSocket = Flux.fromIterable(udpServerProperties.getPorts()).map(port -> {DatagramSocket datagramSocket = vertx.createDatagramSocket(getDatagramSocketOption()).handler(packet -> log.info("【Vertx-UDP-Server】 => 收到数据包:{}", packet.data()));datagramSocket.listen(port, udpServerProperties.getHost()).onComplete(result -> {if (isClosed) {return;}if (result.succeeded()) {log.info("【Vertx-UDP-Server】 => UDP服务启动成功,端口:{}", port);}else {Throwable ex = result.cause();log.error("【Vertx-UDP-Server】 => UDP服务启动失败,错误信息:{}", ex.getMessage(), ex);}});return datagramSocket;});datagramSocket.subscribeOn(Schedulers.boundedElastic()).subscribe();}@Overridepublic synchronized void stop() {isClosed = true;datagramSocket.doOnNext(socket -> socket.close().onComplete(result -> {if (result.succeeded()) {log.info("【Vertx-UDP-Server】 => UDP服务停止成功");}else {Throwable ex = result.cause();log.error("【Vertx-UDP-Server】 => UDP服务停止失败,错误信息:{}", ex.getMessage(), ex);}})).subscribeOn(Schedulers.boundedElastic()).subscribe();}public void deploy() {// 部署服务vertx.deployVerticle(this);// 停止服务Runtime.getRuntime().addShutdownHook(new Thread(this::stop));}private DatagramSocketOptions getDatagramSocketOption() {DatagramSocketOptions datagramSocketOptions = new DatagramSocketOptions();datagramSocketOptions.setBroadcast(udpServerProperties.isBroadcast());datagramSocketOptions.setLoopbackModeDisabled(udpServerProperties.isLoopbackModeDisabled());datagramSocketOptions.setMulticastNetworkInterface(udpServerProperties.getMulticastNetworkInterface());datagramSocketOptions.setIpV6(udpServerProperties.isIpV6());return datagramSocketOptions;}}

VertxUdpServerManager

/*** @author laokou*/
public final class VertxUdpServerManager {private VertxUdpServerManager() {}public static void deploy(final Vertx vertx, final UdpServerProperties properties) {new VertxUdpServer(vertx, properties).deploy();}}
客户端【测试】
/*** @author laokou*/
@Slf4j
@SpringBootTest
@RequiredArgsConstructor
@TestConstructor(autowireMode = TestConstructor.AutowireMode.ALL)
class UdpTest {private final Vertx vertx;@Testvoid test() throws InterruptedException {for (int i = 4880; i < 5000; i++) {DatagramSocket datagramSocket = vertx.createDatagramSocket();int finalI = i;datagramSocket.send("Hello Vertx", i, "127.0.0.1").onComplete(result -> {if (result.succeeded()) {log.info("【Vertx-UDP-Client】 => 发送成功,端口:{}", finalI);}else {Throwable ex = result.cause();log.error("【Vertx-UDP-Client】 => 发送失败,端口:{},错误信息:{}", finalI, ex.getMessage(), ex);}});Thread.sleep(2000);Assertions.assertDoesNotThrow(datagramSocket::close);}}}

这可以满足基本的协议开发,自行修改即可!!!

我是老寇,我们下次再见啦!

http://www.dtcms.com/a/203775.html

相关文章:

  • 【react18】在styled-components中引入图片报错
  • React深度解析:Hooks体系与Redux Toolkit现代状态管理实践
  • Spring Boot集成Spring AI与Milvus实现智能问答系统
  • 广州卓远VR受邀参加2025智能体育典型案例调研活动,并入驻国体华为运动健康联合实验室!
  • 【C/C++】探索单例模式:线程安全与性能优化
  • SQL数据处理流程
  • 微软正式发布 SQL Server 2025 公开预览版,深度集成AI功能
  • PostgreSQL日常维护
  • Flink SQL 计算实时指标同比的实现方法
  • 云原生安全之PaaS:从基础到实践的技术指南
  • 零基础教程:用 Docker + pgloader 将 MySQL 数据库迁移到 PostgreSQL
  • UE5.6新版本—— 动画光照系统重点更新
  • docker 启动一个python环境的项目
  • 详解鸿蒙仓颉开发语言中的日志打印问题
  • linux查看本机服务器的外网IP命令
  • 系分论文《论软件系统安全分析和应用》
  • Web Workers 使用指南
  • 鸿蒙开发:应用上架第二篇,申请发布证书
  • 攻防世界——Web题 fakebook
  • 锐捷网络设备(交换机、路由器、防火墙)常用命令整理
  • 第3周作业-1层隐藏层的神经网络分类二维数据
  • 如何利用 Conda 安装 Pytorch 教程 ?
  • 小土堆pytorch--神经网络-非线性激活线性层及其他层介绍
  • 物联网相关词汇
  • 【JVM】学习笔记
  • API测试框架全解析
  • Java线程池调优与实践经验
  • Tiny C 编译器中,如何实现宏展开和头文件包含的预处理逻辑?
  • maven快速上手
  • iOS 15.4.1 TrollStore(巨魔商店)安装教程详解:第二篇