springboot は mqtt を統合します
クラスターを使用している場合は、セットアップ時に次のポートを開くことを忘れないでください:
#わかりました。構築が成功したら、次のステップは Java プログラムを mqtt に接続することです。接続には 2 つの方法 (実際には 3 つ以上) があります。
1 つは、MQTT Java クライアント ライブラリを直接使用する方法です
2 spring 統合 mqtt
の使用も推奨されており、これに重点を置いています。
最初のステップは、Maven 依存関係を追加することです
<dependency> <groupId>org.springframework.integration</groupId> <artifactId>spring-integration-mqtt</artifactId> <version>5.5.14</version> </dependency>
2 番目のステップは、構成を追加することです
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,断开连接,重连后使用原来的会话 保留订阅的主题,能接收离线期间的消息)
2。次に、対応するクラスを記述しますMqttProperties
import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; /** * MqttProperties * * @author hengzi * @date 2022/8/23 */ @Component public class MqttProperties { /** * 用户名 */ @Value("${mqtt.username}") private String username; /** * 密码 */ @Value("${mqtt.password}") private String password; /** * 连接地址 */ @Value("${mqtt.host-url}") private String hostUrl; /** * 进-客户Id */ @Value("${mqtt.in-client-id}") private String inClientId; /** * 出-客户Id */ @Value("${mqtt.out-client-id}") private String outClientId; /** * 客户Id */ @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; /**是否清除session*/ @Value("${mqtt.clearSession}") private boolean clearSession; // ...getter and setter }
次のステップここには、pipechannel
、adapteradapter
、inboundInbound
、outboundOutbound
などの概念的なものがたくさんあります。など、非常に頭の痛い問題のようです。
わかりました。一つずつやってみましょう。
まず、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; }
次に、送信と受信の 2 つのチューブ (チャネル
) をさらに 2 つ作成します
//出站消息管道, @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; /** * MqttGateway * * @author hengzi * @date 2022/8/23 */ @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; /** * MqttConfig * * @author hengzi * @date 2022/8/23 */ @Configuration public 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
@Component public class MqttMessageHandle implements MessageHandler { @Override public void handleMessage(Message<?> message) throws MessagingException { } }
を処理しています。さらに理解したところ、最適化できる領域があることがわかりました。たとえば、チャネルにはさまざまな種類があります。ここで使用される 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()); }
は、実際にはここで実行します。
しかし、この設定は実際には少し多すぎて少し面倒なので、公式 Web サイトを検索したところ、Java 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; /** * MqttConfigV2 * * @author hengzi * @date 2022/8/24 */ @Configuration public 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(); } }
handleMessage これはメソッドで実行されます,
@Override public void handleMessage(Message<?> message) throws MessagingException { }
も実装できますが、明らかに、多くのトピックを購読していると、書くのが大変になります。これには 2 つのアイデアがあります。問題の 1 つは、さまざまなトピックに従って、
のルーティング router
を別の channel
に追加することです。これが達成できるかどうかもわかっているので、 2 つ目は、名前の変更方法がわかりません。
の @Controller
のデザインを参考にしています。誰もが知っているように、このインターフェイスはすべて @Controller
注釈をクラスに追加します。これは、クラスが http インターフェイスであることを意味します。
をメソッドに追加して、さまざまな URL 呼び出しを実装します。さまざまなメソッド。パラメータ設計では、クラスに
@MqttService
を追加します。これは、このクラスがmqtt メッセージの処理に特化したサービス クラス
@MqttTopic を追加すると、このトピックがこのメソッドによって処理されることを意味します。理論はあります。次のステップは実践です。
最初に 2 つのアノテーションを定義します<div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:java;">import org.springframework.core.annotation.AliasFor;
import org.springframework.stereotype.Component;
import java.lang.annotation.*;
@Documented
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Component
public @interface MqttService {
@AliasFor(
annotation = Component.class
)
String value() default "";
}</pre><div class="contentsignin">ログイン後にコピー</div></div>
Add
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 ""; }
Reference
@RequestMapping次のように使用する必要があります:<div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:java;">import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
/**
* MqttTopicHandle
*
* @author hengzi
* @date 2022/8/24
*/
@MqttService
public 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());
}
}</pre><div class="contentsignin">ログイン後にコピー</div></div>
OK 次のステップは、このような使用方法を実装することです
analytics:メッセージを受信すると、IOC コンテナから
を持つすべてのメッセージが見つかります。アノテーションが付けられたクラス
次に、これらのクラスを走査し、
@MqttTopic# を持つメソッドを見つけます。## を実行し、
@MqttTopic
一貫性がある場合は、このメソッドを実行します
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; /** * MessageHandleService * * @author hengzi * @date 2022/8/24 */ @Component public class MqttMessageHandle implements MessageHandler { public static final Logger log = LoggerFactory.getLogger(MqttMessageHandle.class); // 包含 @MqttService注解 的类(Component) public static Map<String, Object> mqttServices; /** * 所有mqtt到达的消息都会在这里处理 * 要注意这个方法是在线程池里面运行的 * @param message message */ @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); } } } } } } /** * mqtt 订阅的主题与我实际的主题是否匹配 * @param topic 是实际的主题 * @param pattern 是我订阅的主题 可以是通配符模式 * @return 是否匹配 */ 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; } }
Tool class
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; /** * spring工具类 方便在非spring管理环境中获取bean * */ @Component public final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware { /** Spring应用上下文环境 */ 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; } /** * 获取对象 * * @param name * @return Object 一个以所给名字注册的bean的实例 * @throws org.springframework.beans.BeansException * */ @SuppressWarnings("unchecked") public static <T> T getBean(String name) throws BeansException { return (T) beanFactory.getBean(name); } /** * 获取类型为requiredType的对象 * * @param clz * @return * @throws org.springframework.beans.BeansException * */ public static <T> T getBean(Class<T> clz) throws BeansException { T result = (T) beanFactory.getBean(clz); return result; } /** * 如果BeanFactory包含一个与所给名称匹配的bean定义,则返回true * * @param name * @return boolean */ public static boolean containsBean(String name) { return beanFactory.containsBean(name); } /** * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到,将会抛出一个异常(NoSuchBeanDefinitionException) * * @param name * @return boolean * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException * */ public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException { return beanFactory.isSingleton(name); } /** * @param name * @return Class 注册对象的类型 * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException * */ public static Class<?> getType(String name) throws NoSuchBeanDefinitionException { return beanFactory.getType(name); } /** * 如果给定的bean名字在bean定义中有别名,则返回这些别名 * * @param name * @return * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException * */ public static String[] getAliases(String name) throws NoSuchBeanDefinitionException { return beanFactory.getAliases(name); } /** * 获取aop代理对象 * * @param invoker * @return */ @SuppressWarnings("unchecked") public static <T> T getAopProxy(T invoker) { return (T) AopContext.currentProxy(); } /** * 获取当前的环境配置,无配置返回null * * @return 当前的环境配置 */ 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; /** * MqttService * * @author hengzi * @date 2022/8/25 */ @Service public 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 を統合する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。