这篇“springboot如何整合mqtt”文章的知识点大部分人都不太理解,所以小编给大家总结了以下内容,内容详细,步骤清晰,具有一定的借鉴价值,希望大家阅读完这篇文章能有所收获,下面我们一起来看看这篇“springboot如何整合mqtt”文章吧。
springboot 整合 mqtt
搭建的时候如果你使用的是集群 记得开放以下端口:
好了, 搭建成功下一步就是我们的java程序要与mqtt连接, 这里有两种方式(其实不止两种)进行连接.
一是 直接使用 MQTT Java 客户端库
二是使用 spring integration mqtt
也是比较推荐的一种,也是我们主讲这种.
第一步 添加 maven dependency
<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> <version>5.5.14</version> </dependency>
第二步 添加配置
1 先写好一些基本配置
mqtt: username: test # 账号 password: 123456 # 密码 host-url: tcp://127.0.0.1:1883 # mqtt连接tcp地址 in-client-id: ${random.value} # 随机值,使出入站 client ID 不同 out-client-id: ${random.value} client-id: ${random.int} # 客户端Id,不能相同,采用随机数 ${random.value} default-topic: test/#,topic/+/+/up # 默认主题 timeout: 60 # 超时时间 keepalive: 60 # 保持连接 clearSession: true # 清除会话(设置为false,断开连接,重连后使用原来的会话 保留订阅的主题,能接收离线期间的消息)
然后写一个对应的类MqttProperties
import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Component;@Componentpublic class MqttProperties { @Value("${mqtt.username}") private String username; @Value("${mqtt.password}") private String password; @Value("${mqtt.host-url}") private String hostUrl; @Value("${mqtt.in-client-id}") private String inClientId; @Value("${mqtt.out-client-id}") private String outClientId; @Value("${mqtt.client-id}") private String clientId; @Value("${mqtt.default-topic}") private String defaultTopic; @Value("${mqtt.timeout}") private int timeout; @Value("${mqtt.keepalive}") private int keepalive; @Value("${mqtt.clearSession}") private boolean clearSession;// ...getter and setter}
接下来就是配置一些乱七八糟的东西, 这里有很多概念性的东西 比如 管道channel
, 适配器 adapter
, 入站Inbound
, 出站Outbound
,等等等等, 看起来是非常头痛的
好吧,那就一个一个来,
首先连接mqtt需要一个客户端, 那么我们就开一个客户端工厂, 这里可以产生很多很多的客户端
@Bean public MqttPahoClientFactory mqttPahoClientFactory(){ DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); MqttConnectOptions options = new MqttConnectOptions(); options.setServerURIs(mqttProperties.getHostUrl().split(",")); options.setUserName(mqttProperties.getUsername()); options.setPassword(mqttProperties.getPassword().toCharArray()); factory.setConnectionOptions(options); return factory; }
然后再搞两根管子(channel
),一个出站,一个入站
//出站消息管道, @Bean public MessageChannel mqttOutboundChannel(){ return new DirectChannel(); } // 入站消息管道 @Bean public MessageChannel mqttInboundChannel(){ return new DirectChannel(); }
为了使这些管子能流通 就需要一个适配器(adapter
)
// Mqtt 管道适配器 @Bean public MqttPahoMessageDrivenChannelAdapter adapter(MqttPahoClientFactory factory){ return new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getInClientId(),factory,mqttProperties.getDefaultTopic().split(",")); }
然后定义消息生产者
// 消息生产者 @Bean public MessageProducer mqttInbound(MqttPahoMessageDrivenChannelAdapter adapter){ adapter.setCompletionTimeout(5000); adapter.setConverter(new DefaultPahoMessageConverter()); //入站投递的通道 adapter.setOutputChannel(mqttInboundChannel()); adapter.setQos(1); return adapter; }
那我们收到消息去哪里处理呢,答案是这里:
@Bean //使用ServiceActivator 指定接收消息的管道为 mqttInboundChannel,投递到mqttInboundChannel管道中的消息会被该方法接收并执行 @ServiceActivator(inputChannel = "mqttInboundChannel") public MessageHandler handleMessage() { // 这个 mqttMessageHandle 其实就是一个 MessageHandler 的实现类(这个类我放下面) return mqttMessageHandle;// 你也可以这样写// return new MessageHandler() {// @Override// public void handleMessage(Message<?> message) throws MessagingException {// // do something// }// };
到这里我们其实已经可以接受到来自mqtt的消息了
接下来配置向mqtt发送消息
配置 出站处理器
// 出站处理器 @Bean @ServiceActivator(inputChannel = "mqttOutboundChannel") public MessageHandler mqttOutbound(MqttPahoClientFactory factory){ MqttPahoMessageHandler handler = new MqttPahoMessageHandler(mqttProperties.getOutClientId(),factory); handler.setAsync(true); handler.setConverter(new DefaultPahoMessageConverter()); handler.setDefaultTopic(mqttProperties.getDefaultTopic().split(",")[0]); return handler; }
这个 出站处理器 在我看来就是让别人 (MqttPahoMessageHandler
)处理了, 我就不处理了,我只管我要发送什么,至于怎么发送,由MqttPahoMessageHandler
来完成
接下来我们定义一个接口即可
import org.springframework.integration.annotation.MessagingGateway;import org.springframework.integration.mqtt.support.MqttHeaders;import org.springframework.messaging.handler.annotation.Header;import org.springframework.stereotype.Component;@Component@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")public interface MqttGateway { void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String data); void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) Integer Qos, String data);}
我们直接调用这个接口就可以向mqtt 发送数据
到目前为止,整个配置文件长这样:
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.integration.annotation.ServiceActivator;import org.springframework.integration.channel.DirectChannel;import org.springframework.integration.core.MessageProducer;import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;import org.springframework.integration.mqtt.core.MqttPahoClientFactory;import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;import org.springframework.messaging.Message;import org.springframework.messaging.MessageChannel;import org.springframework.messaging.MessageHandler;import org.springframework.messaging.MessagingException;@Configurationpublic class MqttConfig { @Autowired private MqttProperties mqttProperties; //Mqtt 客户端工厂 @Bean public MqttPahoClientFactory mqttPahoClientFactory(){ DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); MqttConnectOptions options = new MqttConnectOptions(); options.setServerURIs(mqttProperties.getHostUrl().split(",")); options.setUserName(mqttProperties.getUsername()); options.setPassword(mqttProperties.getPassword().toCharArray()); factory.setConnectionOptions(options); return factory; } // Mqtt 管道适配器 @Bean public MqttPahoMessageDrivenChannelAdapter adapter(MqttPahoClientFactory factory){ return new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getInClientId(),factory,mqttProperties.getDefaultTopic().split(",")); } // 消息生产者 @Bean public MessageProducer mqttInbound(MqttPahoMessageDrivenChannelAdapter adapter){ adapter.setCompletionTimeout(5000); adapter.setConverter(new DefaultPahoMessageConverter()); //入站投递的通道 adapter.setOutputChannel(mqttInboundChannel()); adapter.setQos(1); return adapter; } // 出站处理器 @Bean @ServiceActivator(inputChannel = "mqttOutboundChannel") public MessageHandler mqttOutbound(MqttPahoClientFactory factory){ MqttPahoMessageHandler handler = new MqttPahoMessageHandler(mqttProperties.getOutClientId(),factory); handler.setAsync(true); handler.setConverter(new DefaultPahoMessageConverter()); handler.setDefaultTopic(mqttProperties.getDefaultTopic().split(",")[0]); return handler; } @Bean //使用ServiceActivator 指定接收消息的管道为 mqttInboundChannel,投递到mqttInboundChannel管道中的消息会被该方法接收并执行 @ServiceActivator(inputChannel = "mqttInboundChannel") public MessageHandler handleMessage() { return mqttMessageHandle; } //出站消息管道, @Bean public MessageChannel mqttOutboundChannel(){ return new DirectChannel(); } // 入站消息管道 @Bean public MessageChannel mqttInboundChannel(){ return new DirectChannel(); }}
处理消息的 MqttMessageHandle
@Componentpublic class MqttMessageHandle implements MessageHandler { @Override public void handleMessage(Message<?> message) throws MessagingException { }}
在进一步了解之后,发现可以优化的地方,比如channel 的类型是有很多种的, 这里使用的DirectChannel
,是Spring Integration
默认的消息通道,它将消息发送给为一个订阅者,然后阻碍发送直到消息被接收,传输方式都是同步的方式,都是由一个线程来运行的.
这里我们可以将入站channel
改成 ExecutorChannel
一个可以使用多线程的channel
@Bean public ThreadPoolTaskExecutor mqttThreadPoolTaskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // 最大可创建的线程数 int maxPoolSize = 200; executor.setMaxPoolSize(maxPoolSize); // 核心线程池大小 int corePoolSize = 50; executor.setCorePoolSize(corePoolSize); // 队列最大长度 int queueCapacity = 1000; executor.setQueueCapacity(queueCapacity); // 线程池维护线程所允许的空闲时间 int keepAliveSeconds = 300; executor.setKeepAliveSeconds(keepAliveSeconds); // 线程池对拒绝任务(无线程可用)的处理策略 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); return executor; } // 入站消息管道 @Bean public MessageChannel mqttInboundChannel(){ // 用线程池 return new ExecutorChannel(mqttThreadPoolTaskExecutor()); }
到这里其实可以运行了.
但是这样配置其实还是有点多, 有点乱, 于是我查找官网, f发现一种更简单的配置方法 叫 Java DSL
我们参考官网,稍微改一下,使用 DSL的方式进行配置:
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.integration.channel.ExecutorChannel;import org.springframework.integration.dsl.IntegrationFlow;import org.springframework.integration.dsl.IntegrationFlows;import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;import org.springframework.integration.mqtt.core.MqttPahoClientFactory;import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.ThreadPoolExecutor;@Configurationpublic class MqttConfigV2 { @Autowired private MqttProperties mqttProperties; @Autowired private MqttMessageHandle mqttMessageHandle; //Mqtt 客户端工厂 所有客户端从这里产生 @Bean public MqttPahoClientFactory mqttPahoClientFactory(){ DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); MqttConnectOptions options = new MqttConnectOptions(); options.setServerURIs(mqttProperties.getHostUrl().split(",")); options.setUserName(mqttProperties.getUsername()); options.setPassword(mqttProperties.getPassword().toCharArray()); factory.setConnectionOptions(options); return factory; } // Mqtt 管道适配器 @Bean public MqttPahoMessageDrivenChannelAdapter adapter(MqttPahoClientFactory factory){ return new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getInClientId(),factory,mqttProperties.getDefaultTopic().split(",")); } // 消息生产者 (接收,处理来自mqtt的消息) @Bean public IntegrationFlow mqttInbound(MqttPahoMessageDrivenChannelAdapter adapter) { adapter.setCompletionTimeout(5000); adapter.setQos(1); return IntegrationFlows.from( adapter) .channel(new ExecutorChannel(mqttThreadPoolTaskExecutor())) .handle(mqttMessageHandle) .get(); } @Bean public ThreadPoolTaskExecutor mqttThreadPoolTaskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); // 最大可创建的线程数 int maxPoolSize = 200; executor.setMaxPoolSize(maxPoolSize); // 核心线程池大小 int corePoolSize = 50; executor.setCorePoolSize(corePoolSize); // 队列最大长度 int queueCapacity = 1000; executor.setQueueCapacity(queueCapacity); // 线程池维护线程所允许的空闲时间 int keepAliveSeconds = 300; executor.setKeepAliveSeconds(keepAliveSeconds); // 线程池对拒绝任务(无线程可用)的处理策略 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); return executor; } // 出站处理器 (向 mqtt 发送消息) @Bean public IntegrationFlow mqttOutboundFlow(MqttPahoClientFactory factory) { MqttPahoMessageHandler handler = new MqttPahoMessageHandler(mqttProperties.getOutClientId(),factory); handler.setAsync(true); handler.setConverter(new DefaultPahoMessageConverter()); handler.setDefaultTopic(mqttProperties.getDefaultTopic().split(",")[0]); return IntegrationFlows.from( "mqttOutboundChannel").handle(handler).get(); }}
这样看起来真的简单多了, 头也没那么大了, 我要是早知道多好.
好了以上就是配置相关的, 到这里其实是已经完成springboot 与 mqtt 的整合了.
但其实我一直有个想法, 就是我们接收的消息 都是在 handleMessage
这个方法里面执行的,
@Override public void handleMessage(Message<?> message) throws MessagingException { }
所以我就有了一个想法, 能不能根据 我订阅的主题,在不同的方法执行, 对于这个问题,其实你用if ... else ...
也能实现, 但很明显,如果我订阅的主题很多的话, 那写起来就很头痛了.
对于这个问题,有两种思路, 一个是添加Spring Integration
的路由 router
,根据不同topic路由到不同的channel
, 这个我也知道能不能实现, 我这里就不讨论了.
第二种是, 我也不知道名字改如何叫, 我是参考了 spring
的 @Controller
的设计, 暂且叫他注解模式.
众所周知,我们的接口都是在类上加 @Controller
这个注解, 就代表这个类是 http 接口, 再在方法加上 @RequestMapping
就能实现不同的 url 调用不同的方法.
参数这个设计 我们在类上面加 @MqttService
就代表这个类是专门处理mqtt消息的服务类
同时 在这个类的方法上 加上 @MqttTopic
就代表 这个主题由这个方法处理.
OK, 理论有了,接下来就是 实践.
先定义 两个注解
import org.springframework.core.annotation.AliasFor;import org.springframework.stereotype.Component;import java.lang.annotation.*;@Documented@Target({ElementType.TYPE})@Retention(RetentionPolicy.RUNTIME)@Componentpublic @interface MqttService { @AliasFor( annotation = Component.class ) String value() default "";}
加上 @Component
注解 spring就会扫描, 并注册到IOC容器里
import java.lang.annotation.ElementType;import java.lang.annotation.Retention;import java.lang.annotation.RetentionPolicy;import java.lang.annotation.Target;@Target(ElementType.METHOD)@Retention(RetentionPolicy.RUNTIME)public @interface MqttTopic { String value() default "";}
参考 @RequestMapping
我们使用起来应该是这样的:
import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.messaging.Message;@MqttServicepublic class MqttTopicHandle { public static final Logger log = LoggerFactory.getLogger(MqttTopicHandle.class);// 这里的 # 号是通配符 @MqttTopic("test/#") public void test(Message<?> message){ log.info("test="+message.getPayload()); }// 这里的 + 号是通配符 @MqttTopic("topic/+/+/up") public void up(Message<?> message){ log.info("up="+message.getPayload()); }// 注意 你必须先订阅 @MqttTopic("topic/1/2/down") public void down(Message<?> message){ log.info("down="+message.getPayload()); }}
OK 接下来就是实现这样的使用
分析 :
当我们收到消息时, 我们从IOC容器中 找到所有 带 @MqttService
注解的类
然后 遍历这些类, 找到带有 @MqttTopic
的方法
接着 把 @MqttTopic
注解的的值 与 接受到的topic 进行对比
如果一致则执行这个方法
废话少说, 上代码
import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.messaging.Message;import org.springframework.messaging.MessageHandler;import org.springframework.messaging.MessagingException;import org.springframework.stereotype.Component;import java.lang.reflect.InvocationTargetException;import java.lang.reflect.Method;import java.util.Map;@Componentpublic class MqttMessageHandle implements MessageHandler { public static final Logger log = LoggerFactory.getLogger(MqttMessageHandle.class); // 包含 @MqttService注解 的类(Component) public static Map<String, Object> mqttServices; @Override public void handleMessage(Message<?> message) throws MessagingException { getMqttTopicService(message); } public Map<String, Object> getMqttServices(){ if(mqttServices==null){ mqttServices = SpringUtils.getBeansByAnnotation(MqttService.class); } return mqttServices; } public void getMqttTopicService(Message<?> message){ // 在这里 我们根据不同的 主题 分发不同的消息 String receivedTopic = message.getHeaders().get("mqtt_receivedTopic",String.class); if(receivedTopic==null || "".equals(receivedTopic)){ return; } for(Map.Entry<String, Object> entry : getMqttServices().entrySet()){ // 把所有带有 @MqttService 的类遍历 Class<?> clazz = entry.getValue().getClass(); // 获取他所有方法 Method[] methods = clazz.getDeclaredMethods(); for ( Method method: methods ){ if (method.isAnnotationPresent(MqttTopic.class)){ // 如果这个方法有 这个注解 MqttTopic handleTopic = method.getAnnotation(MqttTopic.class); if(isMatch(receivedTopic,handleTopic.value())){ // 并且 这个 topic 匹配成功 try { method.invoke(SpringUtils.getBean(clazz),message); return; } catch (IllegalAccessException e) { e.printStackTrace(); log.error("代理炸了"); } catch (InvocationTargetException e) { log.error("执行 {} 方法出现错误",handleTopic.value(),e); } } } } } } public static boolean isMatch(String topic, String pattern){ if((topic==null) || (pattern==null) ){ return false; } if(topic.equals(pattern)){ // 完全相等是肯定匹配的 return true; } if("#".equals(pattern)){ // # 号代表所有主题 肯定匹配的 return true; } String[] splitTopic = topic.split("/"); String[] splitPattern = pattern.split("/"); boolean match = true; // 如果包含 # 则只需要判断 # 前面的 for (int i = 0; i < splitPattern.length; i++) { if(!"#".equals(splitPattern[i])){ // 不是# 号 正常判断 if(i>=splitTopic.length){ // 此时长度不相等 不匹配 match = false; break; } if(!splitTopic[i].equals(splitPattern[i]) && !"+".equals(splitPattern[i])){ // 不相等 且不等于 + match = false; break; } } else { // 是# 号 肯定匹配的 break; } } return match; }}
工具类 SpringUtils
import org.springframework.aop.framework.AopContext;import org.springframework.beans.BeansException;import org.springframework.beans.factory.NoSuchBeanDefinitionException;import org.springframework.beans.factory.config.BeanFactoryPostProcessor;import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;import org.springframework.context.ApplicationContext;import org.springframework.context.ApplicationContextAware;import org.springframework.stereotype.Component;import java.util.Map;@Componentpublic final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware { private static ConfigurableListableBeanFactory beanFactory; private static ApplicationContext applicationContext; public static Map<String, Object> getBeansByAnnotation(Class clsName) throws BeansException{ return beanFactory.getBeansWithAnnotation(clsName); } @Override public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { SpringUtils.beanFactory = beanFactory; } @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { SpringUtils.applicationContext = applicationContext; } @SuppressWarnings("unchecked") public static <T> T getBean(String name) throws BeansException { return (T) beanFactory.getBean(name); } public static <T> T getBean(Class<T> clz) throws BeansException { T result = (T) beanFactory.getBean(clz); return result; } public static boolean containsBean(String name) { return beanFactory.containsBean(name); } public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException { return beanFactory.isSingleton(name); } public static Class<?> getType(String name) throws NoSuchBeanDefinitionException { return beanFactory.getType(name); } public static String[] getAliases(String name) throws NoSuchBeanDefinitionException { return beanFactory.getAliases(name); } @SuppressWarnings("unchecked") public static <T> T getAopProxy(T invoker) { return (T) AopContext.currentProxy(); } public static String[] getActiveProfiles() { return applicationContext.getEnvironment().getActiveProfiles(); }}
OK, 大功告成. 终于舒服了, 终于不用写if...else...
了, 个人感觉这样处理起来会更加优雅. 写代码最重要是什么, 是优雅~
以上!
参考文章:
使用 Spring integration 在Springboot中集成Mqtt
Spring Integration(一)概述
附:
动态添加主题方式:
import org.springframework.beans.factory.annotation.Autowired;import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;import org.springframework.stereotype.Service;import java.util.Arrays;@Servicepublic class MqttService { @Autowired private MqttPahoMessageDrivenChannelAdapter adapter; public void addTopic(String topic) { addTopic(topic, 1); } public void addTopic(String topic,int qos) { String[] topics = adapter.getTopic(); if(!Arrays.asList(topics).contains(topic)){ adapter.addTopic(topic,qos); } } public void removeTopic(String topic) { adapter.removeTopic(topic); }}
直接调用就行
以上就是关于“springboot如何整合mqtt”这篇文章的内容,相信大家都有了一定的了解,希望小编分享的内容对大家有帮助,若想了解更多相关的知识内容,请关注编程网行业资讯频道。