说实话从第三家公司听过和设备间通信是用MQTT的，但是实际上工作几年真没怎么用过，刚好最近做一个边缘盒子就用到了，本来打算用华为开源的KubeEdge的，但是华为的东西一贯坑老多而且没有Java的demo，最后还是用EMQX和mqtt实现了，设备内部的通信，特此记录。
首先大概介绍一下MQTT，你可以把它大致看做一种MQ中间件和Kafka那种差不多，但是MQTT其实一开始是美国IBM公司为卫星通信开发的一种工作在TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，全称MQTT(消息队列遥测传输）。然后我工作过的几家和设备相关的公司里其实大部分同事在和设备通信的过程中一般是建议用MQTT的，当然也有相当一部分为了方便和实时性而选择websocket(我本人是很不建议这么干的，因为websocket的应用场景根本不是为设备间通信设计的，websocket实际上更加适用于前后端交互尤其是大屏上使用，因为它很占服务器资源，一台8核16G的服务器根本撑不住几万个websocket链接，而且它还是有状态的，不方便集群拓展，虽然也有广播等集群拓展方式，但总的来说我个人觉得它不是设备通信的最优解，MQTT才是最优解)。
MQTT作为一个消息中间件那么也有客户端（client）和服务端（broker），这里选几个热门开源的 MQTT Broker，其中部分项目提供商业支持，做简单选型对比。
1.png
2.png
3.png
4.png
这里我用的是EMQ，因为它功能特性较为丰富，社区活跃度也较高，当然用Mosquitto的也比较多，华为的KubeEdge其实用的就是Mosquitto，毕竟Mosquitto比较轻量。
yml文件配置：
mqtt:
#连接地址：写死宿主机的dockerIP
address: tcp://172.17.0.1:31883
#客户端id：随便填，没有会自动生成
clientId: box_test
#用户名
username: emqx_test
#密码
password: emqx_test_password
#会话心跳时间
keepAlive: 60
#超时时间
timeout: 60
#订阅的topic
subtopic: BOX/TEST
mqtt配置Java类：
@Component
@Data
@ConfigurationProperties("mqtt")
public class MqttConfig {
/**
* 连接地址
*/
private String address;
/**
* 客户端id
*/
private String clientId;
/**
* 用户名
*/
private String username;
/**
* 密码
*/
private String password;
/**
* 会话心跳时间
*/
private int keepAlive;
/**
* 超时时间
*/
private int timeout;
/**
* 订阅的消息
*/
private String subTopic;
}
mqtt消息格式：
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class MqttMessage<T> implements Serializable {
/**
* 行为（自定义枚举）
*
* @see MessageAction
*/
private String action;
/**
* 消息类型（自定义枚举）
*
* @see MessageType
*/
private String type;
/**
* 消息数据
*/
private T message;
}
mqtt回调：
@Component
@Slf4j
public class Callback implements MqttCallback {
@Resource
private MqttConfig mqttConfig;
@Resource
private MqttClient mqttClient;
@Override
public void connectionLost(Throwable cause) {
// 连接丢失后，一般在这里面进行重连
log.info("控制中心mqtt断连:", cause);
mqttClient.connect();
}
/**
* 发送消息，消息到达后处理方法
*
* @param token
*/
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
int messageId = token.getMessageId();
String[] topics = token.getTopics();
log.info("消息发送完成,messageId={},topics={}", messageId, Arrays.toString(topics));
}
/**
* 订阅主题接收到消息处理方法
*
* @param topic
* @param message
*/
@Override
public void messageArrived(String topic, MqttMessage message) {
// 根据 topic 名称获得 deviceId
String deviceId = MqttTopic.getDeviceIdByTopic(topic);
MqttClient mqttClient = BeanGetter.getBean(MqttClient.class);
String request = new String(message.getPayload());
log.info("接受到的主题：{}, 接受到的消息：{}, messageId：{}, qos：{}", topic, request, message.getId(), message.getQos());
MessageRequest messageRequest = JSON.parseObject(request, MessageRequest.class);
MessageAction action = messageRequest.getAction();
MessageType type = messageRequest.getType();
Object obj = messageRequest.getMessage();
switch (action) {
case NOTIFY:
switch (type) {
case CONTROL:
//do some thing
break;
case SEARCH_DEVICE_STATUS:
//do some thing
break;
default:
break;
}
break;
case INVITE:
//do some thing
break;
default:
break;
}
}
}
mqtt客户端：
@Component
@Slf4j
public class MqttClient {
public org.eclipse.paho.client.mqttv3.MqttClient cMqttClient;
@Resource
private Callback callback;
public boolean sendMessage(String topic, MqttMessage message) {
try {
cMqttClient.publish(topic, message);
} catch (MqttException e) {
e.printStackTrace();
log.error("服务响应包发送失败 ", e);
return false;
}
return true;
}
public org.eclipse.paho.client.mqttv3.MqttClient getClient() {
return cMqttClient;
}
/**
*无参连接mqtt broker
*/
@PostConstruct
public void connect() {
org.eclipse.paho.client.mqttv3.MqttClient client;
//连接mqtt服务器
try {
client = new org.eclipse.paho.client.mqttv3.MqttClient(mqttConfig.getAddress(), mqttConfig.getClientId(), new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(mqttConfig.getUsername());
options.setPassword(mqttConfig.getPassword().toCharArray());
options.setCleanSession(true);
client.setCallback(innerCallback);
log.info("正在mqtt服务器建立连接: {}", mqttConfig.getAddress());
client.connect(options);
log.info("完成mqtt服务器的连接: {}", mqttConfig.getAddress());
mqttClient = client;
//订阅主题
try {
client.subscribe(mqttConfig.getSubTopic());
log.info("订阅主题成功 inner subscribe: {}", mqttConfig.getSubTopic());
} catch (MqttException e) {
log.error("订阅主题失败 inner subscribe fail: {}", mqttConfig.getSubTopic(), e);
}
} catch (MqttException e) {
e.printStackTrace();
log.error("连接mqtt服务器失败请检查配置与inner-mqtt服务器状态");
}
}
public void reconnect() {
connect();
}
/**
* mqtt 消费确认
*/
public void manualMessageArrivedComplete(Integer messageId, Integer qos) {
try {
cMqttClient.messageArrivedComplete(messageId, qos);
log.info("messageId = {},qos = {} ,消息确认成功！", messageId, qos);
} catch (MqttException e) {
log.error("消息确认失败 messageId={}, qos={}", messageId, qos);
} catch (NullPointerException e) {
log.error("messageId or qos 为空");
}
}
/**
*有参连接mqtt broker
*/
public void mqttClientConnect(CheckMqttResponse checkMqttResponse) {
if(cMqttClient != null){
try {
cMqttClient.disconnect();
//强制取消之前的控制中心mqtt订阅
cMqttClient.close(true);
} catch (MqttException e) {
log.error("控制中心取消订阅失败: ", e);
}
}
//要订阅的topics
List<String> topics = new ArrayList<>();
topics.add(String.format("DEVICE/CLIENT/%s", checkMqttResponse.getClientId()));
topics.add(String.format("DEVICE/SERVER/%s", checkMqttResponse.getClientId()));
org.eclipse.paho.client.mqttv3.MqttClient client;
//连接mqtt服务器
try {
client = new org.eclipse.paho.client.mqttv3.MqttClient(checkMqttResponse.getBroker(), checkMqttResponse.getClientId(), new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(checkMqttResponse.getUsername());
options.setPassword(checkMqttResponse.getPassword().toCharArray());
options.setCleanSession(true);
client.setCallback(callback);
log.info("正在与mqtt服务器建立连接: {}", checkMqttResponse.getBroker());
client.connect(options);
log.info("完成与mqtt服务器的连接: {}", checkMqttResponse.getBroker());
cMqttClient = client;
//订阅主题
try {
for (String topic : topics) {
log.info("订阅主题：{}", topic);
cMqttClient.subscribe(topic, 1);
}
} catch (MqttException e) {
log.error("订阅主题失败: ", e);
}
} catch (MqttException e) {
log.error("连接mqtt服务器失败请检查配置与控制中心mqtt服务器状态", e);
}
}
}
断线重连定时任务(每隔2秒检测一次连接状态)：
@Component
@Slf4j
public class MqttReconnectRunner {
@Resource
private MqttClient mqttClient;
@Scheduled(fixedDelay = 2000)
public void reconnect() {
MqttClient cMqttClient = mqttClient.getClient();
if (cMqttClient != null) {
//mqtt断联
if (!cMqttClient.isConnected()) {
log.info("mqtt断连,重试中...");
mqttClient.connect();
}
} else {
//mqtt未初始化
mqttClient.connect();
}
}
}
