基于paho的MQTT Java SDK.zip

package com.ypp.mqtt.sdk.service.subscribe.impl; import com.ypp.mqtt.sdk.exception.MqttSubscribeException; import com.ypp.mqtt.sdk.service.AbstractMqttClient; import com.ypp.mqtt.sdk.service.subscribe.MqttMessageCallback; import com.ypp.mqtt.sdk.service.subscribe.MqttSubscribe; import org.eclipse.paho.client.mqttv3.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * MQTT订阅消息客户端 * * @author ypp * @version 1.0.0 * @date 2020-04-19 19:41 **/ public class MqttSubscribeClient extends AbstractMqttClient implements MqttSubscribe { private static Logger logger = LoggerFactory.getLogger(MqttSubscribeClient.class); private MqttConnectOptions mqttConnectOptions; @Override public void subscribe(String topic, int qos, MqttMessageCallback mqttMessageCallback) throws MqttSubscribeException { MqttClient mqttClient = super.getMqttClient(); this.buildCallback(mqttClient, new String[]{topic}, new int[]{qos}, mqttMessageCallback); try { mqttClient.connect(this.mqttConnectOptions); System.out.println("成功建立连接"); } catch (MqttException e) { logger.error("MQTT连接发生异常", e); throw new MqttSubscribeException("MQTT连接发生异常", e); } } @Override public void subscribe(String[] topic, int[] qos, MqttMessageCallback mqttMessageCallback) throws MqttSubscribeException { MqttClient mqttClient = super.getMqttClient(); this.buildCallback(mqttClient, topic, qos, mqttMessageCallback); try { mqttClient.connect(this.mqttConnectOptions); } catch (MqttException e) { logger.error("MQTT连接发生异常", e); throw new MqttSubscribeException("MQTT连接发生异常", e); } } private void buildCallback(final MqttClient mqttClient, final String[] topic, final int[] qos, final MqttMessageCallback mqttMessageCallback) { mqttClient.setCallback(new MqttCallbackExtended() { @Override public void connectComplete(boolean b, String s) { System.out.println("连接成功"); logger.debug("MQTT连接成功"); try { mqttClient.subscribe(topic, qos); System.out.println("订阅成功"); } catch (MqttException e) { e.printStackTrace(); logger.error("订阅"+ topic +"出现异常", e); } } @Override public void connectionLost(Throwable throwable) { System.out.println("连接断开"); logger.error("MQTT消费者断开连接", throwable); } @Override public void messageArrived(String s, MqttMessage mqttMessage) throws Exception { logger.debug("收到主题：" + s + "，发送来的消息:" + new String(mqttMessage.getPayload())); mqttMessageCallback.messageArrived(s, mqttMessage); } @Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { System.out.println("delivery complete"); logger.debug("delivery complete" + iMqttDeliveryToken); } }); } @Override public MqttClient buildMqttClientConnection(MqttClient mqttClient, MqttConnectOptions mqttConnectOptions) throws MqttException { this.mqttConnectOptions = mqttConnectOptions; return mqttClient; } }