SpringBoot2.0集成MQTT功能之消息订阅处理
SpringBoot2.0集成MQTT功能之消息订阅处理
⽹上资料还是蛮多的，但也不是很全⾯，⽐如如何设置多个client，如何监听不同topic等，好了，废话不多说，还是跟上篇⼀样的环境，上代码：
第⼀，pom配置，引⼊相关jar:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-stream</artifactId>
妇女节放假吗 国家规定</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
</dependency>
第⼆，配置MQTT服务器基本信息，在springBoot配置⽂件application.properties中配置，添加如下：
#MQTT配置信息
#MQTT-⽤户名
spring.mqtt.username=admin
#MQTT-密码
spring.mqtt.password=password
#MQTT-服务器连接地址，如果有多个，⽤逗号隔开，如：tcp://127.0.0.1:61613，tcp://192.168.2.133:61613
spring.mqtt.url=tcp://127.0.0.1:61613
#MQTT-连接服务器默认客户端ID
spring.mqtt.client.id=mqttId
#MQTT-默认的消息推送主题，实际可在调⽤接⼝时指定
spring.pic=topic
#连接超时
spring.mqttpletionTimeout=3000
第三，配置MQTT消息接收处理类：
/**
* 〈⼀句话功能简述〉<br>
* 〈MQTT接收消息处理〉
qq消息*
* @author lenovo
* @create 2018/6/4
* @since 1.0.0
兽人之单亲记*/
@Configuration
@IntegrationComponentScan
public class MqttReceiveConfig {
@Value("${spring.mqtt.username}")
private String username;
@Value("${spring.mqtt.password}")
private String password;
@Value("${spring.mqtt.url}")
private String hostUrl;
@Value("${spring.mqtt.client.id}")
private String clientId;
@Value("${spring.pic}")
private String defaultTopic;
@Value("${spring.mqttpletionTimeout}")
private int completionTimeout ;  //连接超时
@Bean
public MqttConnectOptions getMqttConnectOptions(){
MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
玉镯子
mqttConnectOptions.setUserName(username);
mqttConnectOptions.CharArray());
mqttConnectOptions.setServerURIs(new String[]{hostUrl});
mqttConnectOptions.setKeepAliveInterval(2);
return mqttConnectOptions;
}
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setConnectionOptions(getMqttConnectOptions());
return factory;
}
//接收通道
@Bean
public MessageChannel mqttInputChannel() {
return new DirectChannel();
}
//配置client，监听的topic
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(clientId+"_inbound", mqttClientFactory(),
"hello","hello1");
adapter.setCompletionTimeout(completionTimeout);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInputChannel());
return adapter;
}
//通过通道获取数据
@Bean
@ServiceActivator(inputChannel = "mqttInputChannel")
public MessageHandler handler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
String topic = Headers().get("mqtt_receivedTopic").toString();
String type = topic.substring(topic.lastIndexOf("/")+1, topic.length());
if("hello".equalsIgnoreCase(topic)){
System.out.println("hello,fuckXX,"+Payload().toString());
}else if("hello1".equalsIgnoreCase(topic)){
System.out.println("hello1,fuckXX,"+Payload().toString());
}
}
};
工作鉴定范文}
}
第四，启动服务测试，使⽤postment调⽤上⼀篇的MQTT发送接⼝，分别往hello,hello1两个topic发送消息，测试接收情况：
由此看出，可以正常监听topic并接收处理消息了。
看到这⾥，朋友们可能有疑问，如果我要配置多个client，应该怎么处理呢？这个也简单，我们只要配置多个通道即可，简单代码如下：
//通道2
@Bean
public MessageChannel mqttInputChannelTwo() {
return new DirectChannel();
}
//配置client2，监听的topic:hell2,hello3
@Bean
public MessageProducer inbound1() {
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter(clientId+"_inboundTwo", mqttClientFactory(),
"hello2","hello3");
adapter.setCompletionTimeout(completionTimeout);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(1);
adapter.setOutputChannel(mqttInputChannelTwo());
return adapter;
}
//通过通道2获取数据
@Bean
@ServiceActivator(inputChannel = "mqttInputChannelTwo")
public MessageHandler handlerTwo() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
String topic = Headers().get("mqtt_receivedTopic").toString();
String type = topic.substring(topic.lastIndexOf("/")+1, topic.length());
if("hello2".equalsIgnoreCase(topic)){
介词后加什么
System.out.println("hello2 clientTwo,"+Payload().toString());
}else if("hello3".equalsIgnoreCase(topic)){
System.out.println("hello3 clientTwo,"+Payload().toString());
}
}
};
}
这样⼀来，我们就配置了两个client，client1监听处理hello、hello1主题消息，client2监听处理hello2、hello3主题，测试⼀下：
从输出结果可以看出，我们发送不同的消息，分别由不同的client处理。所以，⼩伙伴，你理解了吗？
【转载请注明出处——⼤道迷途】

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。