MQTT：
MQ 遥测传输 (MQTT) 是轻量级基于代理的发布/订阅的消息传输协议，设计思想是开放、简单、轻量、易于实现。这些特点使它适用于受限环境。该协议的特点有：
使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合。
一、使用Apollo 搭建MQTT服务器
这里使用的Apollo 1.7.1。
官网：http://activemq.apache.org/index.html
下载地址：http://activemq.apache.org/apollo/download.html
快速开始教程：http://activemq.apache.org/apollo/documentation/getting-started.html
使用Apollo搭建MQTT服务器步骤：
下载Apollo服务器并解压，在CMD环境运行其工作目录下的...\bin\apollo.cmd，命令后面带上参数「create mybroker」，创建服务器实例。这里需要Java环境，系统环境变量下要有JAVA_HOME。
创建实例之后会在bin目录下生成mybroker文件夹，其中 ...\etc\apollo.xml文件下是配置服务器信息的文件，...\etc\users.properties文件包含连接MQTT服务器时用到的用户名和密码，初始默认帐号是admin，密码password；
进入...\mybroker\bin\ 目录，在CMD输入命令「apollo-broker.cmd run」，可以使用TAB键自动补全，运行后输出信息如下：
image.png
其中我们要留意的：
1.阿里云要在安全组配置tcp端口
2.windows服务器配置入站规则
3.MQTT服务器TCP连接端口：tcp://0.0.0.0:61613
4.后台Web管理页面：https://127.0.0.1:61681/或http://127.0.0.1:61680/
登录服务器后，如果MQTT服务器有客户端连接，后台会显示如下
image.png
发送：
public class ServerMQTT {
private static boolean finish=true;
private static Charset utf8 = Charset.forName("utf8");
/**
* 启动入口
*/
public static void main(String[] args) throws MqttException {
final MqttConnect HFMqttConnect = new MqttConnect("tcp://ip:61613","*","*");
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
try {
setMessage(HFMqttConnect);
System.out.println("setMessage");
} catch (MqttException e) {
e.printStackTrace();
}
}
},0,10000);
}
private static void setMessage(MqttConnect mqttConnect1) throws MqttException {
mqttConnect1.message = new MqttMessage();
mqttConnect1.message.setQos(1);
mqttConnect1.message.setRetained(true);//禁用遗嘱
mqttConnect1.publish(mqttConnect1.topic11, mqttConnect1.message);
}}
MqttCOnnect
class MqttConnect {
private MqttClient client;
MqttTopic topic11;
MqttMessage message;
/**
* 构造函数
* @throws MqttException
*/
MqttConnect(String HOST, String clientid, String TOPIC) throws MqttException {
// MemoryPersistence设置clientid的保存形式，默认为以内存保存
client = new MqttClient(HOST, clientid, new MemoryPersistence());
connect(TOPIC);
}
MqttClient getClient() {
return client;
}
private void connect(String TOPIC) {
try {
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("admin");
char mC[]={'p','a','s','s','w','o','r','d'};
options.setPassword(mC);
options.setCleanSession(false);
// 设置超时时间
options.setConnectionTimeout(10);
// 设置会话心跳时间
options.setKeepAliveInterval(20);
client.setCallback(new PushCallback(MqttConnect.this));
client.connect(options);
topic11 = client.getTopic(TOPIC);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
*
* @param topic
* @param message
* @throws MqttPersistenceException
* @throws MqttException
*/
void publish(MqttTopic topic, MqttMessage message) throws MqttPersistenceException, MqttException {
MqttDeliveryToken token = topic.publish(message);
token.waitForCompletion();
System.out.println("message is published completely! " + token.isComplete());
}}
接收：
public class ClientMQTT {
public static final String HOST = "tcp://ip:61613";
public static final String TOPIC = "*";//主题
private static final String clientid = "*";
private MqttClient client;
private MqttConnectOptions options;
private String userName = "admin";
private String passWord = "password";
private ScheduledExecutorService scheduler;
private void start() {
try {
// host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
client = new MqttClient(HOST, clientid, new MemoryPersistence());
// MQTT的连接设置
options = new MqttConnectOptions();
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
options.setCleanSession(true);
// 设置连接的用户名
options.setUserName(userName);
// 设置连接的密码
options.setPassword(passWord.toCharArray());
// 设置超时时间 单位为秒
options.setConnectionTimeout(10);
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
options.setKeepAliveInterval(20);
// 设置回调
client.setCallback(new PushCallback());
MqttTopic topic = client.getTopic(TOPIC);
//setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
options.setWill(topic, "close".getBytes(), 2, true);
client.connect(options);
//订阅消息
int[] Qos = {1};
String[] topic1 = {TOPIC};
client.subscribe(topic1, Qos);
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws MqttException {
ClientMQTT client = new ClientMQTT();
client.start();
}}
回调：
public class PushCallback implements MqttCallback {
private MqttConnect mqttConnect;
PushCallback(MqttConnect mqttConnect) {
this.mqttConnect = mqttConnect;
}
PushCallback() {
}
// MiddlewareMqttClient service;
@Override
public void connectionLost(Throwable cause) {
// 连接丢失后，一般在这里面进行重连
System.out.println("连接断开，可以做重连");
// for (int i = 0; i < 10; i++) {
try {
if (mqttConnect.getClient().isConnected()) {
return;
}
mqttConnect.getClient().reconnect();
} catch (MqttException e) {
e.printStackTrace();
}
// }
throw new RuntimeException("无法连接服务器");
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
// subscribe后得到的消息会执行到这里面
System.out.println("接收消息主题 : " + topic);
System.out.println("接收消息Qos : " + message.getQos());
System.out.println("接收消息内容 : " + new String(message.getPayload()));
}}
image.png
网友评论