package com.fastbee.mq.mqttClient;
|
|
import com.fastbee.common.constant.FastBeeConstant;
|
import com.fastbee.common.core.redis.RedisCache;
|
import com.fastbee.common.enums.FunctionReplyStatus;
|
import com.fastbee.common.exception.ServiceException;
|
import com.fastbee.iot.domain.FunctionLog;
|
import com.fastbee.iot.service.IFunctionLogService;
|
import lombok.extern.slf4j.Slf4j;
|
import org.eclipse.paho.client.mqttv3.*;
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
|
import org.springframework.stereotype.Component;
|
|
import javax.annotation.Resource;
|
|
/**
|
* 发布服务mqtt客户端
|
*/
|
@Component
|
@Slf4j
|
public class PubMqttClient {
|
|
@Resource
|
private MqttClientConfig mqttConfig;
|
@Resource(name = "pubMqttCallBack")
|
private PubMqttCallBack mqttCallBack;
|
/**
|
* 连接配置
|
*/
|
private MqttConnectOptions options;
|
/**
|
* MQTT异步客户端
|
*/
|
private MqttAsyncClient client;
|
/**
|
* 是否连接标记
|
*/
|
private boolean isConnected = false;
|
@Resource
|
private RedisCache redisCache;
|
@Resource
|
private IFunctionLogService functionLogService;
|
|
/**
|
* 启动MQTT客户端
|
*/
|
public synchronized void initialize() {
|
|
try {
|
setOptions();
|
createClient();
|
while (!client.isConnected()) {
|
IMqttToken token = client.connect(options);
|
if(token != null && token.isComplete()) {
|
log.debug("=>内部MQTT客户端启动成功");
|
this.isConnected = true;
|
break;
|
}
|
log.debug("=>内部mqtt客户端连接中...");
|
Thread.sleep(20000);
|
}
|
} catch (MqttException ex) {
|
log.error("=>MQTT客户端初始化异常", ex);
|
} catch (Exception e) {
|
log.error("=>连接MQTT服务器异常", e);
|
this.isConnected = false;
|
}
|
|
}
|
|
public boolean isConnected() {
|
return this.isConnected;
|
}
|
|
private void createClient() {
|
try {
|
if (client == null) {
|
/*host为主机名，clientId是连接MQTT的客户端ID*/
|
client = new MqttAsyncClient(mqttConfig.getHostUrl(), getClientId(), new MemoryPersistence());
|
//设置回调函数
|
client.setCallback(mqttCallBack);
|
mqttCallBack.setClient(client);
|
mqttCallBack.setOptions(this.options);
|
mqttCallBack.setEnabled(mqttConfig.getEnabled());
|
}
|
} catch (Exception e) {
|
log.error("=>mqtt客户端创建错误");
|
}
|
}
|
|
/**
|
* 设置连接属性
|
*/
|
private void setOptions() {
|
|
if (options != null) {
|
options = null;
|
}
|
options = new MqttConnectOptions();
|
options.setConnectionTimeout(mqttConfig.getTimeout());
|
options.setKeepAliveInterval(mqttConfig.getKeepalive());
|
options.setUserName(mqttConfig.getUsername());
|
options.setPassword(mqttConfig.getPassword().toCharArray());
|
//设置自动重新连接
|
options.setAutomaticReconnect(true);
|
/*设置为false，断开连接，不清除session，重连后还是原来的session
|
保留订阅的主题，能接收离线期间的消息*/
|
options.setCleanSession(true);
|
}
|
|
/**
|
* 断开与mqtt的连接
|
*/
|
public synchronized void disconnect() {
|
//判断客户端是否null 是否连接
|
if (client != null && client.isConnected()) {
|
try {
|
IMqttToken token = client.disconnect();
|
token.waitForCompletion();
|
} catch (MqttException e) {
|
log.error("=>断开mqtt连接发生错误 message={}", e.getMessage());
|
throw new ServiceException("断开mqtt连接发生错误" + e.getMessage());
|
}
|
}
|
client = null;
|
}
|
|
/**
|
* 重新连接MQTT
|
*/
|
public synchronized void refresh() {
|
disconnect();
|
initialize();
|
}
|
|
/**
|
* 拼接客户端id
|
*/
|
public final String getClientId() {
|
return FastBeeConstant.SERVER.WM_PREFIX + System.currentTimeMillis();
|
}
|
|
/**
|
* 发布qos=1，非持久化
|
*/
|
public void publish(String topic, byte[] pushMessage, FunctionLog log) {
|
try {
|
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_SEND_TOTAL, -1L);
|
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_SEND_TODAY, 60 * 60 * 24);
|
publish(pushMessage, topic, false, 0);
|
if (null != log) {
|
//存储服务下发成功
|
log.setResultMsg(FunctionReplyStatus.NORELY.getMessage());
|
log.setResultCode(FunctionReplyStatus.NORELY.getCode());
|
functionLogService.insertFunctionLog(log);
|
}
|
} catch (Exception e) {
|
if (null != log) {
|
//服务下发失败存储
|
log.setResultMsg(FunctionReplyStatus.FAIl.getMessage() + "原因: " + e.getMessage());
|
log.setResultCode(FunctionReplyStatus.FAIl.getCode());
|
functionLogService.insertFunctionLog(log);
|
}
|
}
|
}
|
|
/**
|
* 发布主题
|
*
|
* @param message payload消息体
|
* @param topic 主题
|
* @param retained 是否保留消息
|
* @param qos 消息质量
|
* Qos1：消息发送一次，不确保
|
* Qos2：至少分发一次，服务器确保接收消息进行确认
|
* Qos3：只分发一次，确保消息送达和只传递一次
|
*/
|
public void publish(byte[] message, String topic, boolean retained, int qos) {
|
//设置mqtt消息
|
MqttMessage mqttMessage = new MqttMessage();
|
mqttMessage.setQos(qos);
|
mqttMessage.setRetained(retained);
|
mqttMessage.setPayload(message);
|
|
IMqttDeliveryToken token;
|
try {
|
token = client.publish(topic, mqttMessage);
|
token.waitForCompletion();
|
} catch (MqttPersistenceException e) {
|
log.error("=>发布主题时发生错误 topic={},message={}", topic, e.getMessage());
|
throw new ServiceException(e.getMessage());
|
} catch (MqttException ex) {
|
throw new ServiceException(ex.getMessage());
|
}
|
}
|
|
|
/**
|
* 发布
|
*
|
* @param qos 连接方式
|
* @param retained 是否保留
|
* @param topic 主题
|
* @param pushMessage 消息体
|
*/
|
public void publish(int qos, boolean retained, String topic, String pushMessage) {
|
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_SEND_TOTAL, -1L);
|
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_SEND_TODAY, 60 * 60 * 24);
|
log.info("发布主题[{}],发布消息[{}]" + topic,pushMessage);
|
MqttMessage message = new MqttMessage();
|
message.setQos(qos);
|
message.setRetained(retained);
|
message.setPayload(pushMessage.getBytes());
|
|
try {
|
IMqttDeliveryToken token = client.publish(topic, message);
|
token.waitForCompletion();
|
} catch (MqttPersistenceException e) {
|
e.printStackTrace();
|
} catch (MqttException e) {
|
log.error("=>发布主题时发生错误 topic={},message={}", topic, e.getMessage());
|
}
|
}
|
|
}
|
