Java使用ActiveMQ MQTT协议发布消息详解

本文主要介绍了在Java环境中使用ActiveMQ消息中间件，并通过Fusesource库实现MQTT协议来发布消息的三种方法，包括阻塞式（BlockingConnection）、回调式（CallbackConnection）和Future样式（FutureConnection）。Fusesource库的mqtt-client依赖版本为1.12。 首先，我们需要在项目中引入Fusesource提供的mqtt-client依赖，具体Maven配置如下： ```xml <dependency> <groupId>org.fusesource.mqtt-client</groupId> <artifactId>mqtt-client</artifactId> <version>1.12</version> </dependency> ``` 接着，创建MQTT连接并设置主机地址和端口，例如： ```java MQTT mqtt = new MQTT(); mqtt.setHost("localhost", 1883); // 或者 mqtt.setHost("tcp://localhost:1883"); ``` ### 1. 阻塞式连接（BlockingConnection） 阻塞式连接适用于同步操作，它会等待消息的发送和接收完成。以下是如何使用BlockingConnection发布消息和订阅主题的示例： ```java // 建立连接 BlockingConnection connection = mqtt.blockingConnection(); connection.connect(); // 发布消息 connection.publish("消息队列或主题名称", "消息字符串".getBytes(), QoS.AT_LEAST_ONCE, false); // 订阅主题 Topic[] topics = {new Topic("foo", QoS.AT_LEAST_ONCE)}; byte[] qoses = connection.subscribe(topics); String messageString = new String(qoses); // 接收消息 Message message = connection.receive(); System.out.println("队列名称：" + message.getTopic()); byte[] payload = message.getPayload(); String content = new String(payload); // 确认消息消费 message.ack(); // 断开连接 connection.disconnect(); ``` ### 2. 回调式连接（CallbackConnection） 回调式连接适合异步处理，它允许开发者提供一个回调函数来处理消息接收事件。 ```java // 创建回调连接 CallbackConnection callbackConnection = mqtt.callbackConnection(); callbackConnection.connected(new ConnectionFactory() { @Override public void onConnected() { // 连接成功后的操作 } }); callbackConnection.disconnected(new DisconnectedFactory() { @Override public void onDisconnected() { // 断开连接后的操作 } }); callbackConnection.delivered(new DeliveredFactory() { @Override public void onDelivered(String topic, Message message) { // 消息送达时的操作 byte[] payload = message.getPayload(); String content = new String(payload); // 处理消息 } }); // 发布消息 callbackConnection.publish("消息队列或主题名称", "消息字符串".getBytes(), QoS.AT_LEAST_ONCE, false); // 订阅主题 callbackConnection.subscribe(new Topic[]{new Topic("foo", QoS.AT_MOST_ONCE)}); ``` ### 3. Future样式连接（FutureConnection） FutureConnection是基于Future API实现的，它提供了异步操作的接口，可以等待操作完成或者超时。 ```java // 创建FutureConnection FutureConnection futureConnection = mqtt.futureConnection(); // 连接并发布消息 Future<Connection> connectFuture = futureConnection.connect(); connectFuture.sync().publish("消息队列或主题名称", "消息字符串".getBytes(), QoS.AT_LEAST_ONCE, false); // 订阅主题 Future<Future<Byte>> subscribeFuture = futureConnection.subscribe(new Topic[]{new Topic("foo", QoS.AT_MOST_ONCE)}); subscribeFuture.get(); // 接收消息 Future<Message> receiveFuture = futureConnection.receive(); Message message = receiveFuture.get(); System.out.println("队列名称：" + message.getTopic()); byte[] payload = message.getPayload(); String content = new String(payload); // 确认消息消费 message.ack(); // 断开连接 futureConnection.disconnect(); ``` 以上就是使用ActiveMQ的MQTT协议在Java中实现消息发布的三种方式。选择哪种方式取决于你的应用需求，如是否需要同步处理、异步处理，以及对性能和资源管理的考虑。