hivemq/hivemq-mqtt-client

# HiveMQ MQTT Client [](https://maven-badges.herokuapp.com/maven-central/com.hivemq/hivemq-mqtt-client) [](https://javadoc.io/doc/com.hivemq/hivemq-mqtt-client) [](https://img.shields.io/github/actions/workflow/status/hivemq/hivemq-mqtt-client/.github/workflows/check.yml?branch=master) 兼容 MQTT 5.0 和 3.1.1、功能丰富且高性能的 Java 客户端库，提供不同的 API 风格以及 backpressure 支持。 - 文档：https://hivemq.github.io/hivemq-mqtt-client/ - 社区论坛：https://community.hivemq.com/ - HiveMQ 官网：https://www.hivemq.com/ - 贡献指南：[CONTRIBUTING.md](CONTRIBUTING.md) - 许可证：[LICENSE](LICENSE) - MQTT 资源： - [MQTT 基础](https://www.hivemq.com/mqtt-essentials/) - [MQTT 5 基础](https://www.hivemq.com/mqtt-5/) ## 功能 - **所有 [MQTT 3.1.1](http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html) 和 [MQTT 5.0](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html) 功能** - API 风格： - **Reactive**：兼容 [Reactive Streams](https://www.reactive-streams.org/)，提供 [RxJava](https://github.com/ReactiveX/RxJava) 和 [Reactor](https://github.com/reactor/reactor-core) API - **异步 API (Asynchronous API)**：futures 和 callbacks - **阻塞 API (Blocking API)**：用于快速入门和测试 - 灵活切换风格并支持并发使用 - 风格之间划分明确，但保持一致的 API 风格 - **Backpressure 支持**： - QoS 1 和 2 - QoS 0（如有必要，丢弃传入消息） - 将 MQTT 流控制与 reactive 拉取 backpressure 相结合 - 传输协议： - TCP - **SSL/TLS** - 支持最高至 TLS 1.3 的所有 TLS 版本 - TLS 双向认证 - TLS 服务器名称指示 (SNI) - TLS 会话恢复 - 默认且可自定义的主机名验证 - **WebSocket**、安全 WebSocket - **代理 (Proxy)**：SOCKS4、SOCKS5、HTTP CONNECT - 所有可能的组合 - 自动且可配置的 **线程管理** - 自动且可配置的 **重连处理与消息重发** - 如果会话过期，则自动且可配置地 **重新订阅** - **手动消息确认** - 选择性地为特定 stream 启用手动确认 - 对于发射到多个 stream 的消息，在每个 stream 中独立确认 （客户端在发送 MQTT 确认之前会聚合这些确认） - 手动确认的顺序无关紧要 （客户端会自动确保 MQTT 确认的顺序，以 100% 兼容 MQTT 规范） - 生命周期监听器 - 连接建立时 - 断开连接或连接失败时 - MQTT 5 特定功能： - 可插拔的增强认证支持（除了 MQTT 规范外：还包括由服务器触发的重新认证） - 自动 Topic Alias 映射 - 用于 QoS 流的拦截器 ## 用户 [ ](https://github.com/bmwcarit)    [ ](https://github.com/bmwcarit/joynr)    [ ](https://www.openhab.org/)    [ ](https://github.com/eclipse/ditto)    [ ](https://github.com/OSGP/open-smart-grid-platform)    [ ](https://github.com/EXXETA/correomqtt)    [ ](https://github.com/SmartsquareGmbH/mqtt-starter)    [ ](https://www.openmuc.org/)    [ ](https://github.com/leftshiftone/gaia-sdk) 如果您在未列出的项目中使用了 HiveMQ MQTT Client，欢迎提交 issue 或 pull request。 ## 如何使用 需要 Java 8 或更高版本。 ### 依赖 #### Gradle 如果您使用 Gradle，只需在您的 `build.gradle(.kts)` 文件中包含以下内容。 ``` dependencies { implementation("com.hivemq:hivemq-mqtt-client:1.3.15") } ``` 对于可选功能，您可以选择包含额外的模块： ``` dependencies { implementation(platform("com.hivemq:hivemq-mqtt-client-websocket:1.3.15")) implementation(platform("com.hivemq:hivemq-mqtt-client-proxy:1.3.15")) implementation(platform("com.hivemq:hivemq-mqtt-client-epoll:1.3.15")) implementation("com.hivemq:hivemq-mqtt-client-reactor:1.3.15") } ``` #### Maven 如果您使用 Maven，只需在您的 `pom.xml` 文件中包含以下内容。 ``` ... com.hivemq hivemq-mqtt-client 1.3.15 ... ``` 注意：您必须将编译器版本设置为 `1.8` 或更高。 ``` ... 1.8 1.8 ... ``` 对于可选功能，您可以选择包含额外的模块： ``` ... com.hivemq hivemq-mqtt-client-websocket 1.3.15 pom com.hivemq hivemq-mqtt-client-proxy 1.3.15 pom com.hivemq hivemq-mqtt-client-epoll 1.3.15 pom com.hivemq hivemq-mqtt-client-reactor 1.3.15 ... ``` ### Shaded 版本 如果您遇到传递依赖的问题，可以尝试使用 shaded 版本。 此版本将仅在内部使用的传递依赖打包在了一个不同的包名下。 shaded 版本包含了 websocket、proxy 和 epoll 模块。 要使用 shaded 版本，只需在 artifact 名称后追加 `-shaded`。 #### Gradle ``` dependencies { implementation("com.hivemq:hivemq-mqtt-client-shaded:1.3.15") } ``` #### Maven ``` ... com.hivemq hivemq-mqtt-client-shaded 1.3.15 ... ``` ### Snapshots 可以使用 [JitPack](https://jitpack.io/#hivemq/hivemq-mqtt-client) 获取 Snapshots。 #### Gradle ``` repositories { ... maven { url 'https://jitpack.io' } } dependencies { implementation("com.github.hivemq.hivemq-mqtt-client:hivemq-mqtt-client:develop-SNAPSHOT") // snapshots for optional modules implementation(platform("com.github.hivemq.hivemq-mqtt-client:hivemq-mqtt-client-websocket:develop-SNAPSHOT")) implementation(platform("com.github.hivemq.hivemq-mqtt-client:hivemq-mqtt-client-proxy:develop-SNAPSHOT")) implementation(platform("com.github.hivemq.hivemq-mqtt-client:hivemq-mqtt-client-epoll:develop-SNAPSHOT")) implementation("com.github.hivemq.hivemq-mqtt-client:hivemq-mqtt-client-reactor:develop-SNAPSHOT") } ``` #### Maven ``` ... jitpack.io https://jitpack.io com.github.hivemq.hivemq-mqtt-client hivemq-mqtt-client develop-SNAPSHOT com.github.hivemq.hivemq-mqtt-client hivemq-mqtt-client-websocket develop-SNAPSHOT pom com.github.hivemq.hivemq-mqtt-client hivemq-mqtt-client-proxy develop-SNAPSHOT pom com.github.hivemq.hivemq-mqtt-client hivemq-mqtt-client-epoll develop-SNAPSHOT pom com.github.hivemq.hivemq-mqtt-client hivemq-mqtt-client-reactor develop-SNAPSHOT ... ``` 将 artifact 名称更改为 `hivemq-mqtt-client-shaded` 即可获取 shaded 版本的 snapshots。 JitPack 支持所有分支以及特定的 commit。 只需在版本中指定 `-SNAPSHOT` 或 commit id 的前 10 位数字。 ## 基本原则 - API 和实现明确分离。严禁直接使用 `internal` 包中的所有类。 - API 主要是流式的，并使用流式 builder 来创建客户端、配置和消息。 - API 设计保持一致： - 整个库使用相同的原则。 - MQTT 3 和 5 接口尽可能保持一致，仅存在特定版本的差异。 ## 创建客户端 基类：`Mqtt3Client`、`Mqtt5Client` ``` Mqtt5Client client = MqttClient.builder() .identifier(UUID.randomUUID().toString()) .serverHost("broker.hivemq.com") .useMqttVersion5() .build(); Mqtt3Client client = MqttClient.builder()...useMqttVersion3().build(); ``` 或者如果预先知道版本： ``` Mqtt5Client client = Mqtt5Client.builder()...build(); Mqtt3Client client = Mqtt3Client.builder()...build(); ``` 每种 API 风格都有一个特定的 `build...()` 方法。 ## API 风格 每种 API 风格都有各自的接口，以明确区分它们。 随时可以切换 API 风格。 ### 阻塞 API (Blocking API) - Builder 方法：`buildBlocking()` - 切换方法：`client.toBlocking()` #### 示例 ##### 订阅示例 ``` final Mqtt5BlockingClient client = Mqtt5Client.builder() .identifier(UUID.randomUUID().toString()) .serverHost("broker.hivemq.com") .buildBlocking(); client.connect(); try (final Mqtt5Publishes publishes = client.publishes(MqttGlobalPublishFilter.ALL)) { client.subscribeWith().topicFilter("test/topic").qos(MqttQos.AT_LEAST_ONCE).send(); publishes.receive(1, TimeUnit.SECONDS).ifPresent(System.out::println); publishes.receive(100, TimeUnit.MILLISECONDS).ifPresent(System.out::println); } finally { client.disconnect(); } ``` ##### 发布示例 ``` Mqtt5BlockingClient client = Mqtt5Client.builder() .identifier(UUID.randomUUID().toString()) .serverHost("broker.hivemq.com") .buildBlocking(); client.connect(); client.publishWith().topic("test/topic").qos(MqttQos.AT_LEAST_ONCE).payload("1".getBytes()).send(); client.disconnect(); ``` #### 连接 ``` client.connect(); ``` 或者使用自定义属性的 Connect 消息： ``` client.connectWith().keepAlive(10).send(); ``` 或者使用预构建的 Connect 消息： ``` Mqtt5Connect connectMessage = Mqtt5Connect.builder().keepAlive(10).build(); client.connect(connectMessage); ``` #### 发布 ``` client.publishWith() .topic("test/topic") .qos(MqttQos.AT_LEAST_ONCE) .payload("payload".getBytes()) .send(); ``` 或者使用预构建的 Publish 消息： ``` Mqtt5Publish publishMessage = Mqtt5Publish.builder() .topic("test/topic") .qos(MqttQos.AT_LEAST_ONCE) .payload("payload".getBytes()) .build(); client.publish(publishMessage); ``` #### 订阅 ``` client.subscribeWith().topicFilter("test/topic").qos(MqttQos.EXACTLY_ONCE).send(); ``` 或者使用预构建的 Subscribe 消息： ``` Mqtt5Subscribe subscribeMessage = Mqtt5Subscribe.builder() .topicFilter("test/topic") .qos(MqttQos.EXACTLY_ONCE) .build(); client.subscribe(subscribeMessage); ``` #### 取消订阅 ``` client.unsubscribeWith().topicFilter("test/topic").send(); ``` 或者使用预构建的 Unsubscribe 消息： ``` Mqtt5Unsubscribe unsubscribeMessage = Mqtt5Unsubscribe.builder().topicFilter("test/topic").build(); client.unsubscribe(unsubscribeMessage); ``` #### 消费消息 ``` try (Mqtt5BlockingClient.Mqtt5Publishes publishes = client.publishes(MqttGlobalPublishFilter.ALL)) { Mqtt5Publish publishMessage = publishes.receive(); // or with timeout Optional publishMessage = publishes.receive(10, TimeUnit.SECONDS); // or without blocking Optional publishMessage = publishes.receiveNow(); } ``` 必须在 `subscribe` 之前调用 `publishes`，以确保不会丢失任何消息。 也可以在 `connect` 之前调用它，以接收上一个会话的消息。 #### 断开连接 ``` client.disconnect(); ``` 或者使用自定义属性的 DISCONNECT 消息（仅限 MQTT 5）： ``` client.disconnectWith().reasonString("test").send(); ``` 或者使用预构建的 Disconnect 消息（仅限 MQTT 5）： ``` Mqtt5Disconnect disconnectMessage = Mqtt5Disconnect.builder().reasonString("test").build(); client.disconnect(disconnectMessage); ``` #### 重新认证（仅限 MQTT 5） ``` client.reauth(); ``` ### 异步 API (Async API) - Builder 方法：`buildAsync()` - 切换方法：`client.toAsync()` #### 示例 ##### 订阅示例 ``` Mqtt5BlockingClient client = Mqtt5Client.builder() .identifier(UUID.randomUUID().toString()) .serverHost("broker.hivemq.com") .buildBlocking(); client.connect(); client.toAsync().subscribeWith() .topicFilter("test/topic") .qos(MqttQos.AT_LEAST_ONCE) .callback(System.out::println) .send(); ``` ##### 发布示例 ``` Mqtt5AsyncClient client = Mqtt5Client.builder() .identifier(UUID.randomUUID().toString()) .serverHost("broker.hivemq.com") .buildAsync(); client.connect() .thenCompose(connAck -> client.publishWith().topic("test/topic").payload("1".getBytes()).send()) .thenCompose(publishResult -> client.disconnect()); ``` #### 连接 `connect()`、`connectWith()` 和 `connect(Mqtt3/5Connect)` 方法调用与阻塞 API 类似，但返回 `CompletableFuture`。 #### 发布 `publishWith()` 和 `publish(Mqtt3/5Publish)` 方法调用与阻塞 API 类似，但返回 `CompletableFuture`。 #### 订阅 `subscribeWith()` 和 `subscribe(Mqtt3/5Subscribe)` 方法调用与阻塞 API 类似，但返回 `CompletableFuture`。 此外，可以针对每次订阅分别消费消息： ``` client.subscribeWith() .topicFilter("test/topic") .qos(MqttQos.EXACTLY_ONCE) .callback(System.out::println) .executor(executor) // optional .send(); ``` 或者使用预构建的 Subscribe 消息： ``` Mqtt5Subscribe subscribeMessage = Mqtt5Subscribe.builder() .topicFilter("test/topic") .qos(MqttQos.EXACTLY_ONCE) .build(); client.subscribe(subscribeMessage, System.out::println); client.subscribe(subscribeMessage, System.out::println, executor); ``` #### 取消订阅 `unsubscribeWith()` 和 `unsubscribe(Mqtt3/5Unsubscribe)` 方法调用与阻塞 API 类似，但返回 `CompletableFuture`。 #### 消费消息 消息既可以针对每次订阅进行消费（如上所述），也可以全局消费： ``` client.publishes(MqttGlobalPublishFilter.ALL, System.out::println); ``` 或者在指定的 executor 上执行 callback： ``` client.publishes(MqttGlobalPublishFilter.ALL, System.out::println, executor); ``` 必须在 `subscribe` 之前调用 `publishes`，以确保不会丢失任何消息。 也可以在 `connect` 之前调用它，以接收上一个会话的消息。 #### 断开连接 `disconnect()`、`disconnectWith()` 和 `disconnect(Mqtt5Disconnect)` 方法调用与阻塞 API 类似，但 返回 `CompletableFuture`。 #### 重新认证（仅限 MQTT 5） `reauth()` 方法调用与阻塞 API 类似，但返回 `CompletableFuture`。 ### Reactive API - Builder 方法：`buildRx()` - 切换方法：`client.toRx()` #### 示例 ##### 订阅示例 ``` Mqtt5RxClient client = Mqtt5Client.builder() .identifier(UUID.randomUUID().toString()) .serverHost("broker.hivemq.com") .buildRx(); // As we use the reactive API, the following line does not connect yet, but returns a reactive type. // e.g. Single is something like a lazy and reusable future. Think of it as a source for the ConnAck message. Single connAckSingle = client.connect(); // Same here: the following line does not subscribe yet, but returns a reactive type. // FlowableWithSingle is a combination of the single SubAck message and a Flowable of Publish messages. // A Flowable is an asynchronous stream that enables backpressure from the application over the client to the broker. FlowableWithSingle subAckAndMatchingPublishes = client.subscribeStreamWith() .topicFilter("a/b/c").qos(MqttQos.AT_LEAST_ONCE) .addSubscription().topicFilter("a/b/c/d").qos(MqttQos.EXACTLY_ONCE).applySubscription() .applySubscribe(); // The reactive types offer many operators that will not be covered here. // Here we register callbacks to print messages when we received the CONNACK, SUBACK and matching PUBLISH messages. Completable connectScenario = connAckSingle .doOnSuccess(connAck -> System.out.println("Connected, " + connAck.getReasonCode())) .doOnError(throwable -> System.out.println("Connection failed, " + throwable.getMessage())) .ignoreElement(); Completable subscribeScenario = subAckAndMatchingPublishes .doOnSingle(subAck -> System.out.println("Subscribed, " + subAck.getReasonCodes())) .doOnNext(publish -> System.out.println( "Received publish" + ", topic: " + publish.getTopic() + ", QoS: " + publish.getQos() + ", payload: " + new String(publish.getPayloadAsBytes()))) .ignoreElements(); // Reactive types can be easily and flexibly combined connectScenario.andThen(subscribeScenario).blockingAwait(); ``` ##### 发布示例 ``` Mqtt5RxClient client = Mqtt5Client.builder() .identifier(UUID.randomUUID().toString()) .serverHost("broker.hivemq.com") .buildRx(); // As we use the reactive API, the following line does not connect yet, but returns a reactive type. Completable connectScenario = client.connect() .doOnSuccess(connAck -> System.out.println("Connected, " + connAck.getReasonCode())) .doOnError(throwable -> System.out.println("Connection failed, " + throwable.getMessage())) .ignoreElement(); // Fake a stream of Publish messages with an incrementing number in the payload Flowable messagesToPublish = Flowable.range(0, 10_000) .map(i -> Mqtt5Publish.builder() .topic("a/b/c") .qos(MqttQos.AT_LEAST_ONCE) .payload(("test " + i).getBytes()) .build()) // Emit 1 message only every 100 milliseconds .zipWith(Flowable.interval(100, TimeUnit.MILLISECONDS), (publish, i) -> publish); // As we use the reactive API, the following line does not publish yet, but returns a reactive type. Completable publishScenario = client.publish(messagesToPublish) .doOnNext(publishResult -> System.out.println( "Publish acknowledged: " + new String(publishResult.getPublish().getPayloadAsBytes()))) .ignoreElements(); // As we use the reactive API, the following line does not disconnect yet, but returns a reactive type. Completable disconnectScenario = client.disconnect().doOnComplete(() -> System.out.println("Disconnected")); // Reactive types can be easily and flexibly combined connectScenario.andThen(publishScenario).andThen(disconnectScenario).blockingAwait(); ``` #### 连接 `connect()`、`connectWith()` 和 `connect(Mqtt3/5Connect)` 方法调用与异步及阻塞 API 类似，但 返回 `Single`。 #### 发布 `publish` 接收一个包含 Publish 消息的 reactive stream（`Flowable`），并返回一个包含 Publish 结果的 reactive stream （`Flowable`）。 Reactive API 通常不用于发布单个消息。 不过，使用以下代码也是可以实现的。 ``` Single result = client.publish(Flowable.just(Mqtt5Publish.builder() .topic("test/topic") .qos(MqttQos.AT_LEAST_ONCE) .payload("payload".getBytes()) .build())).singleOrError(); ``` #### 订阅 `subscribeWith()` 和 `subscribe(Mqtt3/5Subscribe)` 方法调用与异步及阻塞 API 类似，但返回 `Single`。 此外，可以针对每次订阅分别消费消息： ``` Flowable result = client.subscribeStreamWith() .topicFilter("test/topic") .qos(MqttQos.EXACTLY_ONCE) .applySubscribe() .doOnSingle(subAck -> System.out.println("subscribed")) .doOnNext(publish -> System.out.println("received publish")); ``` 或者使用预构建的 Subscribe 消息： ``` Mqtt5Subscribe subscribeMessage = Mqtt5Subscribe.builder() .topicFilter("test/topic") .qos(MqttQos.EXACTLY_ONCE) .build(); Flowable result = client.subscribeStreamWith(subscribeMessage) .doOnSingle(subAck -> System.out.println("subscribed")) .doOnNext(publish -> System.out.println("received publish")); ``` #### 取消订阅 `unsubscribeWith()` 和 `unsubscribe(Mqtt3/5Unsubscribe)` 方法调用与异步及阻塞 API 类似，但 返回 `Single`。 #### 消费消息 消息既可以针对每次订阅进行消费（如上所述），也可以全局消费： ``` Flowable result = client.publishes(MqttGlobalPublishFilter.ALL).doOnNext(System.out::println); ``` 必须在 `subscribe` 之前调用 `publishes`，以确保不会丢失任何消息。 也可以在 `connect` 之前调用它，以接收上一个会话的消息。 #### 断开连接 `disconnect()`、`disconnectWith()` 和 `disconnect(Mqtt5Disconnect)` 方法调用与异步及阻塞 API 类似，但返回 `Completable`。 #### 重新认证（仅限 MQTT 5） `reauth()` 方法调用与异步及阻塞 API 类似，但返回 `Completable`。 ## 版本管理 使用[语义化版本控制](https://semver.org/)。 严禁直接使用 `com.hivemq.client.internal` 包中的所有代码。它随时可能发生变化，并且不 属于公共 API 的一部分。 严禁实现带有 `DoNotImplement` 注解的接口。其实现由库提供。 这使得库以后可以在不破坏实现类向后兼容性的情况下，向接口添加方法。 ## 贡献 如果您想为 HiveMQ MQTT Client 做出贡献，请参阅[贡献指南](CONTRIBUTING.md)。 ## 许可证 HiveMQ MQTT Client 基于 `APACHE LICENSE, VERSION 2.0` 授权。许可证副本可在[此处](LICENSE)找到。

标签：JS文件枚举, 后台面板检测, 域名枚举, 客户端库, 异步编程, 物联网, 网络通信