物联网之使用Vertx实现MQTT-Server最佳实践【相应式】

欢乐狗  论坛元老 | 2025-5-10 16:39:17 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1970|帖子 1970|积分 5910

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

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

vertx-mqtt地址
实现思路

1.启动MQTT Server并绑定许多端口记录到缓存，服务注册到Nacos，通过接口的方式获取IP和端口【负载均衡】
2.MQTT Client连接MQTT Server并上报数据
3.MQTT Server吸收到数据并通过MQ转发出去
代码比较简单，懒得讲解啦

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

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

实现过程

查看源码
kafka安装

接纳docker-compose一键式启动！！！
还没有安装docker朋友，参考文章下面两篇文章
# Ubuntu20.04安装Docker
# Centos7安装Docker 23.0.6
  1. services:
  2.     kafka:
  3.       image: bitnami/kafka:4.0.0
  4.       container_name: kafka
  5.       tty: true
  6.       ports:
  7.         - '9092:9092'
  8.         - '9093:9093'
  9.       environment:
  10.         # 节点ID
  11.         - KAFKA_BROKER_ID=1
  12.         # 允许使用kraft，即Kafka替代Zookeeper
  13.         - KAFKA_ENABLE_KRAFT=yes
  14.         # kafka角色，做broker，也要做controller
  15.         - KAFKA_CFG_PROCESS_ROLES=broker,controller
  16.         # 指定供外部使用的控制类请求信息
  17.         - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
  18.         # 定义安全协议
  19.         - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
  20.         # 定义kafka服务端socket监听端口
  21.         - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
  22.         # 外网访问地址
  23.         - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
  24.         # 允许使用PLAINTEXT监听器，默认false，不建议在生产环境使用
  25.         - ALLOW_PLAINTEXT_LISTENER=yes
  26.         # 设置broker最大内存，和初始内存
  27.         - KAFKA_HEAP_OPTS=-Xmx512M -Xms512M
  28.         # 集群地址
  29.         - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
  30.         # 节点ID
  31.         - KAFKA_CFG_NODE_ID=1
  32.       restart: always
  33.       privileged: true
  34.       networks:
  35.         - laokou_network
  36. networks:
  37.     laokou_network:
  38.         driver: bridge
复制代码
  1. # 创建topic【进入bin目录执行】 => 每个topic 3个分区和一个副本
  2. kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create --topic laokou_mqtt_property_report --partitions 3 --replication-factor 1
  4. kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create --topic laokou_mqtt_property_reply --partitions 3 --replication-factor 1
复制代码
kafka【相应式】

1.依赖
  1. <dependency>
  2.   <groupId>org.springframework.kafka</groupId>
  3.   <artifactId>spring-kafka</artifactId>
  4.   <version>3.3.5</version>
  5. </dependency>
  6. <dependency>
  7.   <groupId>io.projectreactor.kafka</groupId>
  8.   <artifactId>reactor-kafka</artifactId>
  9.   <version>1.3.23</version>
  10. </dependency>
复制代码
2.代码
KafkaAutoConfig
  1. /**
  2. * @author laokou
  3. */
  4. @Configuration
  5. public class KafkaAutoConfig {
  7.     @Bean("defaultKafkaTemplate")
  8.     @ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.SERVLET)
  9.     public DefaultKafkaTemplate defaultKafkaTemplate(KafkaTemplate<String, String> kafkaTemplate) {
  10.        return new DefaultKafkaTemplate(kafkaTemplate);
  11.     }
  13.     @Bean(value = "reactiveKafkaSender")
  14.     @ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE)
  15.     public KafkaSender reactiveKafkaSender(SenderOptions<String, String> senderOptions) {
  16.        return new ReactiveKafkaSender(
  17.              new reactor.kafka.sender.internals.DefaultKafkaSender<>(ProducerFactory.INSTANCE, senderOptions));
  18.     }
  20.     @Bean
  21.     @ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE)
  22.     public SenderOptions<String, String> senderOptions(KafkaProperties kafkaProperties) {
  23.        Map<String, Object> props = new HashMap<>();
  24.        KafkaProperties.Producer producer = kafkaProperties.getProducer();
  25.        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
  26.        props.put(ProducerConfig.ACKS_CONFIG, producer.getAcks());
  27.        props.put(ProducerConfig.RETRIES_CONFIG, producer.getRetries());
  28.        props.put(ProducerConfig.BATCH_SIZE_CONFIG, (int) producer.getBatchSize().toBytes());
  29.        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, (int) producer.getBufferMemory().toBytes());
  30.        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  31.        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
  32.        return SenderOptions.create(props);
  33.     }
  35.     @Bean
  36.     @ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE)
  37.     public ReceiverOptions<String, String> receiverOptions(KafkaProperties kafkaProperties) {
  38.        Map<String, Object> props = new HashMap<>();
  39.        KafkaProperties.Consumer consumer = kafkaProperties.getConsumer();
  40.        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
  41.        props.put(ConsumerConfig.GROUP_ID_CONFIG, consumer.getGroupId());
  42.        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, consumer.getMaxPollRecords());
  43.        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, consumer.getEnableAutoCommit());
  44.        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  45.        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  46.        return ReceiverOptions.create(props);
  47.     }
  49. }
复制代码
KafkaSender
  1. /**
  2. * @author laokou
  3. */
  4. public interface KafkaSender {
  6.     Flux<Boolean> send(String topic, String payload);
  8. }
复制代码
ReactiveKafkaSender
  1. /**
  2. * @author laokou
  3. */
  4. @Slf4j
  5. @RequiredArgsConstructor
  6. public class ReactiveKafkaSender implements KafkaSender {
  8.     private final DefaultKafkaSender<String, String> defaultKafkaSender;
  10.     @Override
  11.     public Flux<Boolean> send(String topic, String payload) {
  12.        return defaultKafkaSender.send(Mono.just(SenderRecord.create(topic, null, null, null, payload, null)))
  13.           .map(result -> {
  14.              Exception exception = result.exception();
  15.              if (ObjectUtils.isNotNull(exception)) {
  16.                 log.error("【Kafka】 => 发送消息失败，错误信息：{}", exception.getMessage(), exception);
  17.                 return false;
  18.              }
  19.              else {
  20.                 return true;
  21.              }
  22.           });
  23.     }
  25. }
复制代码
3.yaml设置【自动批量提交】
  1. spring:
  2.   kafka:
  3.     bootstrap-servers: kafka:9092
  4.     consumer:
  5.       group-id: laokou-mqtt
  6.       # 禁用自动提交（按周期）已消费offset
  7.       enable-auto-commit: true
  8.       # 单次poll()调用返回的记录数
  9.       max-poll-records: 50
  10.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  11.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  12.     producer:
  13.       # 发生错误后，消息重发的次数。
  14.       retries: 5
  15.       # 当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。
  16.       batch-size: 16384
  17.       # 设置生产者内存缓冲区的大小。
  18.       buffer-memory: 33554432
  19.       # 键的序列化方式
  20.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  21.       # 值的序列化方式
  22.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  23.       # acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
  24.       # acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
  25.       # acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
  26.       acks: 0
  27.     listener:
  28.       # 在侦听器容器中运行的线程数。
  29.       concurrency: 5
  30.       # 批量提交模式
  31.       ack-mode: batch
  32.       # 批量batch类型
  33.       type: batch
  34.       # topic不存在报错
  35.       missing-topics-fatal: false
  36.     admin:
  37.       auto-create: false
复制代码
mqtt-server【相应式】

依赖
  1. <dependency>
  2.   <groupId>io.vertx</groupId>
  3.   <artifactId>vertx-mqtt</artifactId>
  4.   <version>4.5.14</version>
  5. </dependency>
复制代码
VertxConfig
  1. /**
  2. * @author laokou
  3. */
  4. @Configuration
  5. public class VertxConfig {
  7.     @Bean(destroyMethod = "close")
  8.     public Vertx vertx() {
  9.        VertxOptions vertxOptions = new VertxOptions();
  10.        vertxOptions.setMaxEventLoopExecuteTime(30);
  11.        vertxOptions.setWorkerPoolSize(40);
  12.        vertxOptions.setMaxWorkerExecuteTime(30);
  13.        vertxOptions.setMaxEventLoopExecuteTimeUnit(TimeUnit.SECONDS);
  14.        vertxOptions.setMaxWorkerExecuteTimeUnit(TimeUnit.SECONDS);
  15.        vertxOptions.setPreferNativeTransport(true);
  16.        vertxOptions.setInternalBlockingPoolSize(40);
  17.        vertxOptions.setEventLoopPoolSize(Math.max(32, 2 * CpuCoreSensor.availableProcessors()));
  18.        return Vertx.vertx(vertxOptions);
  19.     }
  21. }
复制代码
MqttServerProperties【设置了账号和暗码】
  1. /**
  2. * @author laokou
  3. */
  4. @Data
  5. @Component
  6. @ConfigurationProperties(prefix = "spring.mqtt-server")
  7. public class MqttServerProperties {
  9.     private boolean auth = true;
  11.     private String username = "vertx";
  13.     private String password = "laokou123";
  15.     private String host = "0.0.0.0";
  17.     private int port = 0;
  19.     private int threadSize = 32;
  21.     private int maxMessageSize = 8196;
  23.     private boolean isAutoClientId = true;
  25.     private int maxClientIdLength = 30;
  27.     private int timeoutOnConnect = 90;
  29.     private boolean useWebSocket = false;
  31.     private int webSocketMaxFrameSize = 65536;
  33.     private boolean perFrameWebSocketCompressionSupported = true;
  35.     private boolean perMessageWebSocketCompressionSupported = true;
  37.     private int webSocketCompressionLevel = 6;
  39.     private boolean webSocketAllowServerNoContext = false;
  41.     private boolean webSocketPreferredClientNoContext = false;
  43.     private boolean tcpNoDelay = true;
  45.     private boolean tcpKeepAlive = false;
  47.     private int tcpKeepAliveIdleSeconds = -1;
  49.     private int tcpKeepAliveCount = -1;
  51.     private int tcpKeepAliveIntervalSeconds = -1;
  53.     private int soLinger = -1;
  55.     private int idleTimeout = 0;
  57.     private int readIdleTimeout = 0;
  59.     private int writeIdleTimeout = 0;
  61.     private TimeUnit idleTimeoutUnit = TimeUnit.SECONDS;
  63.     private boolean ssl = false;
  65.     private boolean tcpFastOpen = false;
  67.     private boolean tcpCork = false;
  69.     private boolean tcpQuickAck = false;
  71.     private int tcpUserTimeout = 0;
  73. }
复制代码
VertxMqttServer
  1. /**
  2. * @author laokou
  3. */
  4. @Slf4j
  5. public final class VertxMqttServer {
  7.     private final Sinks.Many<MqttPublishMessage> messageSink = Sinks.many()
  8.        .multicast()
  9.        .onBackpressureBuffer(Integer.MAX_VALUE, false);
  11.     private volatile Flux<MqttServer> mqttServer;
  13.     private final Vertx vertx;
  15.     private final MqttServerProperties properties;
  17.     private final List<ReactiveMessageHandler> reactiveMessageHandlers;
  19.     private volatile boolean isClosed = false;
  21.     public VertxMqttServer(final Vertx vertx, final MqttServerProperties properties,
  22.           List<ReactiveMessageHandler> reactiveMessageHandlers) {
  23.        this.properties = properties;
  24.        this.vertx = vertx;
  25.        this.reactiveMessageHandlers = reactiveMessageHandlers;
  26.     }
  28.     public Flux<MqttServer> start() {
  29.        return mqttServer = getMqttServerOptions().map(mqttServerOption -> MqttServer.create(vertx, mqttServerOption)
  30.           .exceptionHandler(
  31.                 error -> log.error("【Vertx-MQTT-Server】 => MQTT服务启动失败，错误信息：{}", error.getMessage(), error))
  32.           .endpointHandler(endpoint -> Optional.ofNullable(authHandler(endpoint))
  33.              .ifPresent(e -> e.closeHandler(close -> log.info("【Vertx-MQTT-Server】 => MQTT客户端断开连接"))
  34.                 .subscribeHandler(subscribe -> {
  35.                    for (MqttTopicSubscription topicSubscription : subscribe.topicSubscriptions()) {
  36.                       log.info("【Vertx-MQTT-Server】 => MQTT客户端订阅主题：{}", topicSubscription.topicName());
  37.                    }
  38.                 })
  39.                 .disconnectHandler(disconnect -> log.info("【Vertx-MQTT-Server】 => MQTT客户端主动断开连接"))
  40.                 .pingHandler(ping -> log.info("【Vertx-MQTT-Server】 => MQTT客户端发送心跳"))
  41.                 .publishHandler(messageSink::tryEmitNext)
  42.                 // 不保留会话
  43.                 .accept(false)))
  44.           .listen(mqttServerOption.getPort(), mqttServerOption.getHost(), asyncResult -> {
  45.              if (isClosed) {
  46.                 return;
  47.              }
  48.              if (asyncResult.succeeded()) {
  49.                 log.info("【Vertx-MQTT-Server】 => MQTT服务启动成功，主机：{}，端口：{}", mqttServerOption.getHost(),
  50.                       mqttServerOption.getPort());
  51.                 // 写入缓存
  52.                 PortCache.add(mqttServerOption.getPort());
  53.              }
  54.              else {
  55.                 log.error("【Vertx-MQTT-Server】 => MQTT服务启动失败，主机：{}，端口：{}，错误信息：{}", mqttServerOption.getHost(),
  56.                       mqttServerOption.getPort(), asyncResult.cause().getMessage(), asyncResult.cause());
  57.              }
  58.           }));
  59.     }
  61.     public Flux<MqttServer> stop() {
  62.        isClosed = true;
  63.        return mqttServer.doOnNext(server -> server.close(completionHandler -> {
  64.           if (completionHandler.succeeded()) {
  65.              log.info("【Vertx-MQTT-Server】 => MQTT服务停止成功");
  66.           }
  67.           else {
  68.              log.error("【Vertx-MQTT-Server】 => MQTT服务停止失败，错误信息：{}", completionHandler.cause().getMessage(),
  69.                    completionHandler.cause());
  70.           }
  71.        }));
  72.     }
  74.     public Flux<Boolean> publish() {
  75.        return messageSink.asFlux().flatMap(message -> {
  76.           // @formatter:off
  77.              // log.info("【Vertx-MQTT-Server】 => MQTT服务接收到消息，主题：{}，内容：{}", message.topicName(), message.payload().toString());
  78.              // @formatter:on
  79.           return Flux
  80.              .fromStream(reactiveMessageHandlers.stream()
  81.                 .filter(reactiveMessageHandler -> reactiveMessageHandler.isSubscribe(message.topicName())))
  82.              .flatMap(reactiveMessageHandler -> reactiveMessageHandler
  83.                 .handle(new MqttMessage(message.payload(), message.topicName())));
  84.        });
  85.     }
  87.     private int detectAvailablePort(String host) {
  88.        try (ServerSocket socket = SSLServerSocketFactory.getDefault().createServerSocket()) {
  89.           socket.bind(new InetSocketAddress(host, properties.getPort()));
  90.           return socket.getLocalPort();
  91.        }
  92.        catch (IOException e) {
  93.           throw new RuntimeException("Port auto-detection failed", e);
  94.        }
  95.     }
  97.     private Flux<MqttServerOptions> getMqttServerOptions() {
  98.        return Flux.range(1, Math.max(properties.getThreadSize(), CpuCoreSensor.availableProcessors()))
  99.           .map(item -> getMqttServerOption());
  100.     }
  102.     /**
  103.      * 认证.
  104.      */
  105.     private MqttEndpoint authHandler(MqttEndpoint endpoint) {
  106.        MqttAuth mqttAuth = endpoint.auth();
  107.        if (properties.isAuth()) {
  108.           if (ObjectUtils.isNull(mqttAuth)) {
  109.              endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USERNAME_OR_PASSWORD);
  110.              return null;
  111.           }
  112.           if (!ObjectUtils.equals(mqttAuth.getUsername(), properties.getUsername())
  113.                 || !ObjectUtils.equals(mqttAuth.getPassword(), properties.getPassword())) {
  114.              endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USERNAME_OR_PASSWORD);
  115.              return null;
  116.           }
  117.        }
  118.        return endpoint;
  119.     }
  121.     // @formatter:off
  122.     private MqttServerOptions getMqttServerOption() {
  123.        MqttServerOptions mqttServerOptions = new MqttServerOptions();
  124.        mqttServerOptions.setHost(properties.getHost());
  125.        mqttServerOptions.setPort(detectAvailablePort(properties.getHost()));
  126.        mqttServerOptions.setMaxMessageSize(properties.getMaxMessageSize());
  127.        mqttServerOptions.setAutoClientId(properties.isAutoClientId());
  128.        mqttServerOptions.setMaxClientIdLength(properties.getMaxClientIdLength());
  129.        mqttServerOptions.setTimeoutOnConnect(properties.getTimeoutOnConnect());
  130.        mqttServerOptions.setUseWebSocket(properties.isUseWebSocket());
  131.        mqttServerOptions.setWebSocketMaxFrameSize(properties.getWebSocketMaxFrameSize());
  132.        mqttServerOptions.setPerFrameWebSocketCompressionSupported(properties.isPerFrameWebSocketCompressionSupported());
  133.        mqttServerOptions.setPerMessageWebSocketCompressionSupported(properties.isPerMessageWebSocketCompressionSupported());
  134.        mqttServerOptions.setWebSocketCompressionLevel(properties.getWebSocketCompressionLevel());
  135.        mqttServerOptions.setWebSocketAllowServerNoContext(properties.isWebSocketAllowServerNoContext());
  136.        mqttServerOptions.setWebSocketPreferredClientNoContext(properties.isWebSocketPreferredClientNoContext());
  137.        mqttServerOptions.setTcpNoDelay(properties.isTcpNoDelay());
  138.        mqttServerOptions.setTcpKeepAlive(properties.isTcpKeepAlive());
  139.        mqttServerOptions.setTcpKeepAliveIdleSeconds(properties.getTcpKeepAliveIdleSeconds());
  140.        mqttServerOptions.setTcpKeepAliveIntervalSeconds(properties.getTcpKeepAliveIntervalSeconds());
  141.        mqttServerOptions.setTcpKeepAliveCount(properties.getTcpKeepAliveCount());
  142.        mqttServerOptions.setSoLinger(properties.getSoLinger());
  143.        mqttServerOptions.setIdleTimeout(properties.getIdleTimeout());
  144.        mqttServerOptions.setReadIdleTimeout(properties.getReadIdleTimeout());
  145.        mqttServerOptions.setWriteIdleTimeout(properties.getWriteIdleTimeout());
  146.        mqttServerOptions.setIdleTimeoutUnit(properties.getIdleTimeoutUnit());
  147.        mqttServerOptions.setSsl(properties.isSsl());
  148.        mqttServerOptions.setTcpFastOpen(properties.isTcpFastOpen());
  149.        mqttServerOptions.setTcpCork(properties.isTcpCork());
  150.        mqttServerOptions.setTcpQuickAck(properties.isTcpQuickAck());
  151.        mqttServerOptions.setTcpUserTimeout(properties.getTcpUserTimeout());
  152.        return mqttServerOptions;
  153.     }
  154.     // @formatter:on
  156. }
复制代码
PortCache【缓存端口】
  1. /**
  2. * @author laokou
  3. */
  4. public final class PortCache {
  6.     private PortCache() {
  7.     }
  9.     public static final List<Integer> PORTS = new CopyOnWriteArrayList<>();
  11.     public static void add(int port) {
  12.        PORTS.add(port);
  13.     }
  15.     public static List<Integer> get() {
  16.        return PORTS;
  17.     }
  19.     public static void clear() {
  20.        PORTS.clear();
  21.     }
  23. }
复制代码
ReactiveMessageHandler【消息处理，没啥好说的，就是用来转发消息到MQ】
  1. /**
  2. * @author laokou
  3. */
  4. public interface ReactiveMessageHandler {
  6.     boolean isSubscribe(String topic);
  8.     Flux<Boolean> handle(MqttMessage mqttMessage);
  10. }
复制代码
  1. /**
  2. * 属性回复消息处理器.
  3. *
  4. * @author laokou
  5. */
  6. @Component
  7. @RequiredArgsConstructor
  8. public class ReactivePropertyReplyMessageHandler implements ReactiveMessageHandler {
  10.     private final KafkaSender kafkaSender;
  12.     @Override
  13.     public boolean isSubscribe(String topic) {
  14.        return TopicUtils.match("/+/+/property/reply", topic);
  15.     }
  17.     @Override
  18.     public Flux<Boolean> handle(MqttMessage mqttMessage) {
  19.        return kafkaSender.send(LAOKOU_MQTT_PROPERTY_REPLY, mqttMessage.getPayload().toString());
  20.     }
  22. }
复制代码
  1. /**
  2. * 属性上报消息处理.
  3. *
  4. * @author laokou
  5. */
  6. @Component
  7. @RequiredArgsConstructor
  8. public class ReactivePropertyReportMessageHandler implements ReactiveMessageHandler {
  10.     private final KafkaSender kafkaSender;
  12.     @Override
  13.     public boolean isSubscribe(String topic) {
  14.        return TopicUtils.match("/+/+/property/report", topic);
  15.     }
  17.     @Override
  18.     public Flux<Boolean> handle(MqttMessage mqttMessage) {
  19.        return kafkaSender.send(LAOKOU_MQTT_PROPERTY_REPORT, mqttMessage.getPayload().toString());
  20.     }
  22. }
复制代码
设置yaml
  1. spring:
  2.   application:
  3.     name: ${SERVICE_ID:laokou-mqtt}
  4.   threads:
  5.     virtual:
  6.       enabled: true
  7.   mqtt-server:
  8.     auth: true
  9.     username: vertx
  10.     password: laokou123
  11.     # 开启8196个端口
  12.     thread-size: 8196
复制代码
启动MQTT-Server
  1. /**
  2. * @author laokou
  3. */
  4. @Slf4j
  5. @EnableDiscoveryClient
  6. @RequiredArgsConstructor
  7. @EnableConfigurationProperties
  8. @SpringBootApplication(scanBasePackages = "org.laokou")
  9. public class MqttServerApp implements CommandLineRunner {
  11.     private final Vertx vertx;
  13.     private final MqttServerProperties properties;
  15.     private final List<ReactiveMessageHandler> reactiveMessageHandlers;
  17.     private final ExecutorService virtualThreadExecutor;
  19.     @Override
  20.     public void run(String... args) {
  21.        virtualThreadExecutor.execute(this::listenMessage);
  22.     }
  24.     private void listenMessage() {
  25.        VertxMqttServer vertxMqttServer = new VertxMqttServer(vertx, properties, reactiveMessageHandlers);
  26.        // 启动服务
  27.        vertxMqttServer.start().subscribeOn(Schedulers.boundedElastic()).subscribe();
  28.        // 推送数据
  29.        vertxMqttServer.publish().subscribeOn(Schedulers.boundedElastic()).subscribe();
  30.        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
  31.           // 清除缓存
  32.           PortCache.clear();
  33.           // 停止服务
  34.           vertxMqttServer.stop().subscribeOn(Schedulers.boundedElastic()).subscribe();
  35.        }));
  36.     }
  38. }
复制代码
启动好之后，请自行测试，这个东西没啥好说，vertx帮我们都实现了，就是简单调用API，自己玩吧~
我是老寇，我们下次再见啦~

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

正序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

欢乐狗

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

渠道 国产数据库 集成商 AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表