简单的demo版

话不多说直接上代码:

mqtt学习文档 https://docs.emqx.io/tutorial/v3/cn/quick_start/whats_mqtt.html

pom.xml

  <!--mqtt--><!-- MQTT-jar --><dependency><groupId>org.eclipse.paho</groupId><artifactId>org.eclipse.paho.client.mqttv3</artifactId><version>1.2.0</version></dependency><dependency><groupId>org.fusesource.mqtt-client</groupId><artifactId>mqtt-client</artifactId><version>1.14</version></dependency>

redis集成pom

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><!-- redis的驱动包:jedis --><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId></dependency>

---

以下就是代码实现

application.properties 配置 redis使用spring 默认的所有没有配置

spring.mqtt.username=
spring.mqtt.password=
spring.mqtt.url=tcp://ip:1883#接收消息主题
spring.mqtt.subscribe=mqtt/subscribe/#
#发布消息主题
spring.mqtt.publish=mqtt/publish/

MqttConfig.java  连接配置类

@Component
@Getter
@Setter
public class MqttConfig {@Value("${spring.mqtt.username}")private String userName; //用户名@Value("${spring.mqtt.password}")private String password;//连接密码@Value("${spring.mqtt.url}")private String url;//连接地址 tcp@Value("${spring.mqtt.subscribe}")private String subscribe; //订阅主题@Value("${spring.mqtt.publish}")private String publish;//发布主题}

这里实现ApplicationRunner 启动加载类 实现run方法 客服端初始化 和连接服务端
注意为什么我这里要把redisTemplate当成参数传入SubMqttMessageCallback类
是因为我们的回调类是不交给spring容器管理的 所以在回调类中 直接@Autowired是无法注入的
每一个客户端连接服务端 都有唯一的clientId

/*** mqtt订阅消息*/
@Component
public class MqttConsumer implements ApplicationRunner {private MqttClient client;@Autowiredprivate MqttConfig config;@Autowiredprivate StringRedisTemplate redisTemplate;@Overridepublic void run(ApplicationArguments args) throws Exception {//初始化连接this.connect();}/*** 连接服务器*/private void connect() throws MqttException {//生成唯一标识String clientId="client"+System.currentTimeMillis();System.out.println(config.getUrl()+config.getUserName()+config.getPassword()+config.getSubscribe());client=new MqttClient(config.getUrl(),clientId, new MemoryPersistence());//初始化连接配置if (client!=null) {MqttConnectOptions options = new MqttConnectOptions();options.setUserName(config.getUserName());options.setPassword(config.getPassword().toCharArray());options.setCleanSession(true);//每次都清除sessionoptions.setConnectionTimeout(3000);//超时时间//设置心跳会话时间options.setKeepAliveInterval(20);try {//注入回调类client.setCallback(new SubMqttMessageCallback(client, options, config.getSubscribe(),redisTemplate,config));client.connect(options);//订阅主题client.subscribe(config.getSubscribe(), 0);//连接成功System.out.println("连接成功 订阅的主题是===>>>" + config.getSubscribe());} catch (MqttException e) {System.out.println("连接失败");e.printStackTrace();}}}
}

MqttCallback 回调类 一共有四个方法 
connectionLost() 用于客户端断线重连  
messageArrived() 订阅消息的处理方法 所有的业务就在这个方法中写

package dges.tech.callback;
/**ProjectName: mqtt*FileName: MqttMessageCallback*UserName: thj*CreateTime:2020/6/16 15:31*/import dges.tech.config.MqttConfig;
import lombok.Getter;
import lombok.Setter;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.data.redis.core.StringRedisTemplate;/*** 消息回调处理*/@Getter
@Setter
public class SubMqttMessageCallback implements MqttCallback {private MqttConnectOptions options;private String topic;private MqttClient client;private MqttConfig config;private StringRedisTemplate redisTemplate;public SubMqttMessageCallback(MqttClient client, MqttConnectOptions options,String topic,StringRedisTemplate redisTemplate,MqttConfig config) {this.client = client;this.options = options;this.topic = topic;this.redisTemplate=redisTemplate;this.config=config;}public SubMqttMessageCallback() {}@Overridepublic void connectionLost(Throwable cause) {//断开重连while (true){try {Thread.sleep(3000);//断开重连client.connect(options);client.subscribe(config.getSubscribe(),0);} catch (InterruptedException | MqttException e) {e.printStackTrace();}}}@Overridepublic void messageArrived(String topic, MqttMessage message) throws Exception {System.out.println(topic);System.out.println(message.getId());String messgaes= new String(message.getPayload(),"UTF-8");System.out.println(messgaes);System.out.println(message.getQos());//发布消息到redis mqtt通道redisTemplate.convertAndSend("mqtt",messgaes);}@Overridepublic void deliveryComplete(IMqttDeliveryToken token) {}
}

---

这里是redis 的发布订阅功能实现了 不想打太多文字 仔细看一下注解吧 嘻嘻 

package dges.tech.listener;
/**ProjectName: mqtt*FileName: RedisMessageListener*UserName: thj*CreateTime:2020/6/16 19:59*/import dges.tech.handler.RedisHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;/*** redis 实现监听*/@Configuration
public class RedisMessageListener {/*** redis消息监听器容器* 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器* 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理* @param connectionFactory* @param listenerAdapter* @return*/@BeanRedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter  listenerAdapter){RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);//监听通道 可以监听多个通道container.addMessageListener(listenerAdapter,new PatternTopic("mqtt"));return container;}/*** 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法* @param handler* @return*/@BeanMessageListenerAdapter adapter(RedisHandler handler){//这个地方 是给messageListenerAdapter 传入一个消息接受的处理器，利用反射的方法调用“receiveMessage”return new MessageListenerAdapter(handler,"receiveMessage");}
}

redis 消息通道处理类

package dges.tech.handler;
/**ProjectName: mqtt*FileName: RedisHandler*UserName: thj*CreateTime:2020/6/16 20:10*/import dges.tech.config.MqttConfig;
import dges.tech.init.MqttConsumer;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;/*** 消息处理器*/
@Component
public class RedisHandler {@Autowiredprivate MqttConsumer consumer;@Autowiredprivate MqttConfig config;/*** 消息处理* @param message*/public void receiveMessage(String message) throws MqttException {System.out.println(message);MqttMessage mqttMessage = new MqttMessage();mqttMessage.setQos(0);mqttMessage.setPayload("handler".getBytes());consumer.publish(mqttMessage,config.getPublish()+message);System.out.println("发送成功");}}

最后看一下运行结果

2020-08-24 15:37:52.438  INFO 11584 --- [           main] dges.tech.MqttApplication                : Starting MqttApplication on LAPTOP-TUITGC1J with PID 11584 (F:\hellomqtt\mqtt\target\classes started by lenovo in F:\hellomqtt\mqtt)
2020-08-24 15:37:52.443  INFO 11584 --- [           main] dges.tech.MqttApplication                : No active profile set, falling back to default profiles: default
2020-08-24 15:37:53.923  INFO 11584 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Multiple Spring Data modules found, entering strict repository configuration mode!
2020-08-24 15:37:53.927  INFO 11584 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Bootstrapping Spring Data repositories in DEFAULT mode.
2020-08-24 15:37:54.152  INFO 11584 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 201ms. Found 0 repository interfaces.
2020-08-24 15:37:55.073  INFO 11584 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat initialized with port(s): 8088 (http)
2020-08-24 15:37:55.123  INFO 11584 --- [           main] o.apache.catalina.core.StandardService   : Starting service [Tomcat]
2020-08-24 15:37:55.123  INFO 11584 --- [           main] org.apache.catalina.core.StandardEngine  : Starting Servlet engine: [Apache Tomcat/9.0.17]
2020-08-24 15:37:55.370  INFO 11584 --- [           main] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring embedded WebApplicationContext
2020-08-24 15:37:55.370  INFO 11584 --- [           main] o.s.web.context.ContextLoader            : Root WebApplicationContext: initialization completed in 2859 ms
2020-08-24 15:37:56.164  INFO 11584 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-08-24 15:37:56.667  INFO 11584 --- [    container-1] io.lettuce.core.EpollProvider            : Starting without optional epoll library
2020-08-24 15:37:56.669  INFO 11584 --- [    container-1] io.lettuce.core.KqueueProvider           : Starting without optional kqueue library
2020-08-24 15:37:59.497  INFO 11584 --- [           main] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8088 (http) with context path ''
2020-08-24 15:37:59.501  INFO 11584 --- [           main] dges.tech.MqttApplication                : Started MqttApplication in 7.719 seconds (JVM running for 10.727)
连接成功 订阅的主题是===>>>mqtt/subscribe/#

mqtt 订阅发布的功能到这里就结束啦  我是一个无情的农码  无情哈拉少