我正在尝试构建一个订阅多个 mqtt 主题的应用程序,获取信息,处理它并形成 xml,并在处理时触发一个事件,以便这些可以发送到某个云服务器,并将成功的响应从那里发送回MQTT 通道。
<int-mqtt:message-driven-channel-adapter
id="mqttAdapter" client-id="${clientId}" url="${brokerUrl}" topics="${topics}"
channel="startCase" auto-startup="true" />
<int:channel id="startCase" />
<int:service-activator id="startCaseService"
input-channel="startCase" ref="msgPollingService" method="pollMessages" />
<bean id="mqttTaskExecutor"
class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="5" />
<property name="maxPoolSize" value="10" />
</bean>
<bean id="msgPollingService" class="com.xxxx.xxx.mqttclient.mqtt.MsgPollingService">
<property name="taskExecutor" ref="mqttTaskExecutor" />
<property name="vendorId" value="${vendorId}" />
</bean>
我的问题是如何将其发布到多个渠道,即我是否可以选择将 X 消息发布到 Y 主题。目前我有以下内容:
<int:channel id="outbound" />
<int-mqtt:outbound-channel-adapter
id="mqtt-publish" client-id="kj" client-factory="clientFactory"
auto-startup="true" url="${brokerUrl}" default-qos="0"
default-retained="true" default-topic="${responseTopic}" channel="outbound" />
<bean id="eventListner" class="com.xxxx.xxxx.mqttclient.event.EventListener">
<property name="sccUrl" value="${url}" />
<property name="restTemplate" ref="restTemplate" />
<property name="channel" ref="outbound" />
</bean>
我可以这样发布:
channel.send(MessageBuilder.withPayload("customResponse").build());
我可以做类似的事情吗:
channel.send(Message<?>, topic)
你的配置看起来不错。但是,那MessageChannel
是松散耦合的抽象,仅处理Message
.
所以,你请求a-lachannel.send(Message<?>, topic)
对于消息传递概念来说不正确。
不过,我们有一个技巧给你。从AbstractMqttMessageHandler
:
String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC);
.....
this.publish(topic == null ? this.defaultTopic : topic, mqttMessage, message);
因此,您的代码需要的是:
channel.send(MessageBuilder.withPayload("customResponse").setHeader(MqttHeaders.TOPIC, topic).build());
换句话说,您应该发送一个Message
with mqtt_topic
标头以实现动态发布<int-mqtt:outbound-channel-adapter>
.
从另一方面我们不建议使用MessageChannel
直接来自应用程序。这<gateway>
带有服务接口就是针对最终应用程序的这种情况。那个地方topic
可以是标记为的服务方法参数之一@Header(MqttHeaders.TOPIC)
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)