|  |  |  | 
|---|
|  |  |  | import lombok.extern.slf4j.Slf4j; | 
|---|
|  |  |  | import org.eclipse.paho.client.mqttv3.*; | 
|---|
|  |  |  | import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; | 
|---|
|  |  |  | import org.springframework.beans.factory.annotation.Value; | 
|---|
|  |  |  | import org.springframework.boot.ApplicationArguments; | 
|---|
|  |  |  | import org.springframework.boot.ApplicationRunner; | 
|---|
|  |  |  | import org.springframework.core.annotation.Order; | 
|---|
|  |  |  | import org.springframework.stereotype.Component; | 
|---|
|  |  |  | import java.io.UnsupportedEncodingException; | 
|---|
|  |  |  | /** | 
|---|
|  |  |  | 
|---|
|  |  |  | **/ | 
|---|
|  |  |  | @Component | 
|---|
|  |  |  | @Slf4j(topic = "mqtt") | 
|---|
|  |  |  | @Order(1) | 
|---|
|  |  |  | public class MqttConsumer implements ApplicationRunner { | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Value("${mqtt.enabled}") | 
|---|
|  |  |  | private Boolean MQTT_ENABLED; | 
|---|
|  |  |  | @Value("${mqtt.topic}") | 
|---|
|  |  |  | private String MQTT_TOPIC; | 
|---|
|  |  |  | @Value("${mqtt.host}") | 
|---|
|  |  |  | private String MQTT_HOST; | 
|---|
|  |  |  | @Value("${mqtt.clientId}") | 
|---|
|  |  |  | private String MQTT_CLIENT_ID; | 
|---|
|  |  |  | @Value("${mqtt.username}") | 
|---|
|  |  |  | private String MQTT_USER_NAME; | 
|---|
|  |  |  | @Value("${mqtt.password}") | 
|---|
|  |  |  | private String MQTT_PASSWORD; | 
|---|
|  |  |  | @Value("${mqtt.timeout}") | 
|---|
|  |  |  | private int MQTT_TIMEOUT; | 
|---|
|  |  |  | @Value("${mqtt.keepalive}") | 
|---|
|  |  |  | private int MQTT_KEEP_ALIVE; | 
|---|
|  |  |  |  | 
|---|
|  |  |  | private static MqttClient client; | 
|---|
|  |  |  |  | 
|---|
|  |  |  |  | 
|---|
|  |  |  | @Override | 
|---|
|  |  |  | public void run(ApplicationArguments args) { | 
|---|
|  |  |  | log.debug("初始化并启动mqtt......"); | 
|---|
|  |  |  | if(PropertiesUtil.MQTT_ENABLED) | 
|---|
|  |  |  | if(MQTT_ENABLED) | 
|---|
|  |  |  | { | 
|---|
|  |  |  | this.connect(); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | 
|---|
|  |  |  | getClient(); | 
|---|
|  |  |  | // 2 设置配置 | 
|---|
|  |  |  | MqttConnectOptions options = getOptions(); | 
|---|
|  |  |  | String[] topic = PropertiesUtil.MQTT_TOPIC.split(","); | 
|---|
|  |  |  | String[] topic = MQTT_TOPIC.split(","); | 
|---|
|  |  |  | // 3 消息发布质量 | 
|---|
|  |  |  | int[] qos = getQos(topic.length); | 
|---|
|  |  |  | // 4 最后设置 | 
|---|
|  |  |  | 
|---|
|  |  |  | public void getClient() { | 
|---|
|  |  |  | try { | 
|---|
|  |  |  | if (null == client) { | 
|---|
|  |  |  | client = new MqttClient(PropertiesUtil.MQTT_HOST, PropertiesUtil.MQTT_CLIENT_ID, new MemoryPersistence()); | 
|---|
|  |  |  | client = new MqttClient(MQTT_HOST, MQTT_CLIENT_ID, new MemoryPersistence()); | 
|---|
|  |  |  | } | 
|---|
|  |  |  | log.debug("--创建mqtt客户端"); | 
|---|
|  |  |  | } catch (Exception e) { | 
|---|
|  |  |  | 
|---|
|  |  |  | public MqttConnectOptions getOptions() { | 
|---|
|  |  |  | MqttConnectOptions options = new MqttConnectOptions(); | 
|---|
|  |  |  | //设置用户名密码 | 
|---|
|  |  |  | options.setUserName(PropertiesUtil.MQTT_USER_NAME); | 
|---|
|  |  |  | options.setPassword(PropertiesUtil.MQTT_PASSWORD.toCharArray()); | 
|---|
|  |  |  | options.setUserName(MQTT_USER_NAME); | 
|---|
|  |  |  | options.setPassword(MQTT_PASSWORD.toCharArray()); | 
|---|
|  |  |  | // 设置超时时间 | 
|---|
|  |  |  | options.setConnectionTimeout(PropertiesUtil.MQTT_TIMEOUT); | 
|---|
|  |  |  | options.setConnectionTimeout(MQTT_TIMEOUT); | 
|---|
|  |  |  | // 设置会话心跳时间 | 
|---|
|  |  |  | options.setKeepAliveInterval(PropertiesUtil.MQTT_KEEP_ALIVE); | 
|---|
|  |  |  | options.setKeepAliveInterval(MQTT_KEEP_ALIVE); | 
|---|
|  |  |  | // 是否清除session | 
|---|
|  |  |  | options.setCleanSession(true); | 
|---|
|  |  |  | log.debug("--生成mqtt配置对象"); | 
|---|