你不知道的vertx mqtt（服务端）

Posted 2021-04-12 互联网步行街

>介绍

mqtt服务端也叫消息代理（broker），位于消息发布者和订阅者之间，主要用来接收来自客户端的连接，接收客户发布的应用消息，处理来自客户的订阅和退订请求，向订阅的客户转发应用程序消息等等

>MQTT协议中定义了一些方法，用来表示对确定资源进行操作。这个资源可以表示预先存在的数据或动态生成数据，这取决于服务器的实现。通常来说，资源指服务器上的文件或输出。

1.connect,等待与服务器简历连接（ 默认使用1883端口启动服务端）

MqttServer mqttServer = MqttServer.create(Vertx.vertx());
mqttServer.endpointHandler(endpoint -> {
    // shows main connect info
    System.out.println("MQTT client [" + endpoint.clientIdentifier()
            + "] request to connect, clean session = "
            + endpoint.isCleanSession());
    if (endpoint.auth() != null) {
        System.out.println("[username = " + endpoint.auth().getUsername()
                + ", password = " + endpoint.auth().getPassword() + "]");
    }
    if (endpoint.will() != null) {
        System.out.println("[will topic = " + endpoint.will().getWillTopic()
                + " msg = " + new String(endpoint.will().getWillMessageBytes()) +
                " QoS = " + endpoint.will().getWillQos()
                + " isRetain = " + endpoint.will().isWillRetain() + "]");
    }
    System.out.println("[keep alive timeout = "
            + endpoint.keepAliveTimeSeconds() + "]");
    // accept connection from the remote client
    endpoint.accept(false);
}).listen(ar -> {
    if (ar.succeeded()) {
        System.out.println("MQTT server is listening on port "
                + ar.result().actualPort());
    } else {
        System.out.println("Error on starting the server");
        ar.cause().printStackTrace();
    }
});

2.Disconnect，等待MQTT客户端完成所做的工作，并于服务器断开tcp/ip会话

endpoint.disconnectHandler(v -> {
    System.out.println("receive disconnect from client");
});

3.Subscribe,等待完成订阅

4.UnSubscribe,等待服务器取消客户端的一个或多个topics订阅

endpoint.unsubscribeHandler(unsubscribe -> {
    for (String t: unsubscribe.topics()) {
        System.out.println("Unsubscription for " + t);
    }
    // ack the subscriptions request
    endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
});

5.Publish。MQTT客户端发送消息请求，发送完成后返回应用程序线程

endpoint.publishHandler(message -> {
    if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
        endpoint.publishAcknowledge(message.messageId());
    } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
        endpoint.publishReceived(message.messageId());
    }
}).publishReleaseHandler(messageId -> {
    endpoint.publishComplete(messageId);
});

最后，给大家一点经济上的建议，美股昨晚熔断，上一次熔断发生在97年。油价掉了22%，已经很接近深海油和页岩油气的开采成本价，这个价格也同时告诉我们实体经济有多差。美国VIX指数也到62，金融海啸后最高位。还有美国企业债券7月份到期，有很大可能会无法偿还。更大的风险还在后面，今年的6，7月份应该会很精彩