MQTT (mosquitto)环境搭建
一、下载mosquitto 安装
二、启动mosquitto
C:\Program Files (x86)\mosquitto>mosquitto.exe -c mosquitto.conf
三、模拟订阅客户端
org.fusesource.mqtt-client
mqtt-client
1.0
package com.suning.ebuy.passbook;
import java.net.URISyntaxException;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
*
* 使用Java 为MQ Telemetry Transport 创建订户在此任务中,您将遵循教程来创建订户应用程序。订户将针对主题创建预订并接收该预订的发布。提供了一个示例订户应用程序Subscribe。Subscribe
* 将创建预订主题MQTT Examples,并等待获得该预订的发布,等待时间为30 秒。订户可以创建预订并等待获得发布。它还可以接收发送至先前为同一客户机标识创建的预订的发布。*
* @author longgangbai
*/
public class WSMQTTClientSubscribe {
private static final Logger LOG = LoggerFactory.getLogger(WSMQTTClientSubscribe.class);
private final static String CONNECTION_STRING = "tcp://localhost:1883";
private final static boolean CLEAN_START = true;
private final static short KEEP_ALIVE = 30;// 低耗网络,但是又需要及时获取数据,心跳30s
private final static String CLIENT_ID = "publishService";
public static Topic[] topics1 = { new Topic("china/beijing", QoS.EXACTLY_ONCE),
new Topic("china/tianjin", QoS.AT_LEAST_ONCE), new Topic("china/henan", QoS.AT_MOST_ONCE) };
public final static long RECONNECTION_ATTEMPT_MAX = 6;
public final static long RECONNECTION_DELAY = 2000;
public final static int SEND_BUFFER_SIZE = 2 * 1024 * 1024;// 发送最大缓冲为2M
public static void main(String[] args) {
// 创建MQTT对象
MQTT mqtt = new MQTT();
BlockingConnection connection = null;
try {
// 设置mqtt broker的ip和端口
mqtt.setHost(CONNECTION_STRING);
// 连接前清空会话信息
mqtt.setCleanSession(CLEAN_START);
// 设置重新连接的次数
mqtt.setReconnectAttemptsMax(RECONNECTION_ATTEMPT_MA X);
// 设置重连的间隔时间
mqtt.setReconnectDelay(RECONNECTION_DELAY);
// 设置心跳时间
mqtt.setKeepAlive(KEEP_ALIVE);
// 设置缓冲的大小
mqtt.setSendBufferSize(SEND_BUFFER_SIZE);
mqtt.setClientId("u2");
// 获取mqtt的连接对象BlockingConnection
connection = mqtt.blockingConnection();
// MQTT连接的创建
connection.connect();
// 创建相关的MQTT 的主题列表
Topic[] topics = { new Topic("hello/world", QoS.AT_LEAST_ONCE) };
// 订阅相关的主题信息
byte[] qoses = connection.subscribe(topics);
//
while (true) {
// 接收订阅的消息内容
Message message = connection.receive();
// 获取订阅的消息内容
byte[] payload = message.getPayload();
// process the message then:
LOG.info("MQTTClient Message Topic=" + message.getTopic() + " Content :" + new String(payload));
// 签收消息的回执
message.ack();
Thread.sleep(2000);
}
} catch (URISyntaxException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
} finally {
try {
connection.disconnect();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
四、发布消息
C:\Program Files (x86)\mosquitto>mosquitto_pub -t hello/world -m dfdfdfdfe2 -i u2
注:发布时可指定订阅者(-i),如不指定,则群发。
浙公网安备 33021202002483