我的理解是,利用第三方EMQ系统,代替了JAVA自己整合EMQ系统框架系统,直接整合后利用API去完成 发布,订阅等操作

1. 在pom文件下添加以下maven依赖：

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-stream</artifactId>
</dependency>
<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId>
</dependency>

2. 在yml文件中进行mqtt的连接配置

spring: mqtt: username: admin	# 账号 password: public # 密码 host-url: tcp://ip:1883 # mqtt连接tcp地址 client-id: test	# 客户端Id，每个启动的id要不同 default-topic: test # 默认主题 timeout: 100 # 超时时间 keepalive: 100	# 保持连接数

3.获取配置

package com.modules.common.mqtt;
import lombok.Getter;
import lombok.Setter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
/** * @Classname MtqqEntity * @Description mqtt相关配置信息 * @Date 2019/4/11 23:00 * @Created by Jack */
@Component
@ConfigurationProperties("spring.mqtt")
@Setter
@Getter
public class MqttConfig { @Autowired private MqttPushClient mqttPushClient; /** * 用户名 */ private String username; /** * 密码 */ private String password; /** * 连接地址 */ private String hostUrl; /** * 客户Id */ private String clientId; /** * 默认连接话题 */ private String defaultTopic; /** * 超时时间 */ private int timeout; /** * 保持连接数 */ private int keepalive; @Bean public MqttPushClient getMqttPushClient() { mqttPushClient.connect(hostUrl, clientId, username, password, timeout, keepalive); // 以/#结尾表示订阅所有以test开头的主题 mqttPushClient.subscribe("test/#", 0); return mqttPushClient; }
}

4.mqtt推送客户端

package com.modules.common.mqtt;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/** * @Classname MqttPushClient * @Description mqtt推送客户端 * @Date 2019/4/11 23:22 * @Created by Jack */
@Component
public class MqttPushClient { private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class); @Autowired private PushCallback pushCallback; private static MqttClient client; private static MqttClient getClient() { return client; } private static void setClient(MqttClient client) { MqttPushClient.client = client; } /** * 客户端连接 * * @param host ip+端口 * @param clientID 客户端Id * @param username 用户名 * @param password 密码 * @param timeout 超时时间 * @param keepalive 保留数 */ public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) { MqttClient client; try { client = new MqttClient(host, clientID, new MemoryPersistence()); MqttConnectOptions options = new MqttConnectOptions(); options.setCleanSession(true); options.setUserName(username); options.setPassword(password.toCharArray()); options.setConnectionTimeout(timeout); options.setKeepAliveInterval(keepalive); MqttPushClient.setClient(client); try { client.setCallback(pushCallback); client.connect(options); } catch (Exception e) { e.printStackTrace(); } } catch (Exception e) { e.printStackTrace(); } } /** * 发布 * * @param qos 连接方式 * @param retained 是否保留 * @param topic 主题 * @param pushMessage 消息体 */ public void publish(int qos, boolean retained, String topic, String pushMessage) { MqttMessage message = new MqttMessage(); message.setQos(qos); message.setRetained(retained); message.setPayload(pushMessage.getBytes()); MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic); if (null == mTopic) { logger.error("topic not exist"); } MqttDeliveryToken token; try { token = mTopic.publish(message); token.waitForCompletion(); } catch (MqttPersistenceException e) { e.printStackTrace(); } catch (MqttException e) { e.printStackTrace(); } } /** * 订阅某个主题 * * @param topic 主题 * @param qos 连接方式 */ public void subscribe(String topic, int qos) { logger.info("开始订阅主题" + topic); try { MqttPushClient.getClient().subscribe(topic, qos); } catch (MqttException e) { e.printStackTrace(); } }
}

5.消费监听类

package com.modules.common.mqtt;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/** * @Classname PushCallback * @Description 消费监听类 * @Date 2019/4/11 23:31 * @Created by Jack */
@Component
public class PushCallback implements MqttCallback { private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class); @Autowired private MqttConfig mqttConfig; private static MqttClient client; @Override public void connectionLost(Throwable throwable) { // 连接丢失后，一般在这里面进行重连 logger.info("连接断开，可以做重连"); if (client == null || !client.isConnected()) { mqttConfig.getMqttPushClient(); } } @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { // subscribe后得到的消息会执行到这里面 logger.info("接收消息主题 : " + topic); logger.info("接收消息Qos : " + mqttMessage.getQos()); logger.info("接收消息内容 : " + new String(mqttMessage.getPayload())); } @Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { logger.info("deliveryComplete---------" + iMqttDeliveryToken.isComplete()); }
}

6.测试

package com.test.controller;
import com.modules.common.exception.ExceptionUtils;
import com.modules.common.mqtt.MqttPushClient;
import com.modules.common.utils.RUtils;
import com.modules.common.validator.ValidatorUtils;
import com.test.entity.TestUser;
import com.test.form.TestForm;
import com.test.service.TestService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.*;
import javax.validation.Valid;
/** * @author: lxw * @Date: 2018/10/19 19:36 * @email: 1229703575@qq.com * @Description: 测试文件 */
@RestController
@RequestMapping("/")
@Api("测试")
public class TestController { @Autowired private MqttPushClient mqttPushClient; @ApiOperation(value = "发布主题", notes = "测试发布主题") @GetMapping(value = "/publishTopic") public RUtils publishTopic() { mqttPushClient.publish(0,false,"test/test","测试一下发布消息"); return RUtils.ok(); }
}

测试成果

1. 进入后台订阅“test/test”主题，注意要进行连接。
   
2. 接口发送相关信息。
   
3. 后台接收到相应的信息。
   
   此时项目中的客户端也订阅了test/下的主题，因此也会收到相应的消息。