引包
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
<version>5.1.3.RELEASE</version>
</dependency>
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.0</version>
</dependency>
————————————————————————————————————————代碼
@Configuration
public class W1_sub {
@Autowired
private MqttConfig mqttConfig;
private String topic= "water/smartt200_up";
//private String topic="sm_xfc_topic";
@Bean
public MqttPahoClientFactory mqttClientFactory_w1() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setServerURIs(mqttConfig.getServers());
return factory;
}
//****************************************************w1_0
@Bean
public MessageChannel iotMqttInputChannel() {
return new DirectChannel();
// return new PublishSubscribeChannel();
}
@Bean
public MessageProducer inbound() {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(MqttAsyncClient.generateClientId(), mqttClientFactory_w1(), topic);
adapter.setCompletionTimeout(5000);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(0);
adapter.setOutputChannel(iotMqttInputChannel());
return adapter;
}
@Bean
@ServiceActivator(inputChannel = "iotMqttInputChannel")
public MessageHandler handlerTest() {
return message -> {
try {
String string = message.getPayload().toString();
System.out.println("w1_1：" + string);
} catch (MessagingException ex) {
//logger.info(ex.getMessage());
}
};
}
//————————————————————————————
@Bean
public MessageChannel iotMqttInputChannel_w1_2() {
return new DirectChannel();
// return new PublishSubscribeChannel();
}
@Bean
public MessageProducer inbound_w1_2() {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(MqttAsyncClient.generateClientId(), mqttClientFactory_w1(), topic);
adapter.setCompletionTimeout(5000);
adapter.setConverter(new DefaultPahoMessageConverter());
adapter.setQos(0);
adapter.setOutputChannel(iotMqttInputChannel_w1_2());
return adapter;
}
@Bean
@ServiceActivator(inputChannel = "iotMqttInputChannel_w1_2")
public MessageHandler handler_w1_2() {
return message -> {
try {
String string = message.getPayload().toString();
System.out.println("W1_2：" + string);
} catch (MessagingException ex) {
//logger.info(ex.getMessage());
}
};
}
}
——————————————————————————————————————上報結果資料
w1_1：{
"dev_name": "Csmartt200",
"varible": {
"DCD4.DCSW_WD0_R0": "0.000000",
"DCD4.DCSW_WD0_R1": "429961.000000",
"DCD4.DCSW_WD0_R2": "0.000000",
"DCD4.DCSW_WD0_R3": "0.000000",
"DCD4.DCSW_WD0_R4": "0.000000",
"DCD4.DCSW_WD0_R5": "194956.000000",
"DCD4.DCSW_WD0_R6": "0.000000",
"DCD4.DCSW_WD0_R7": "3.000000",
"DCD4.DCSW_WD0_R8": "0.000000",
"DCD4.DCSW_WD0_R9": "0.000000",
"DCD4.DCSW_WD0_R10": "0.000000",
"DCD4.DCSW_WD0_R11": "0.000000",
"DCD4.DCSW_WD0_R12": "0.000000",
"DCD4.DCSW_WD0_R13": "1.000000",
"DCD4.DCSW_WD0_R14": "171524.000000"
},
"up_time": "20210301124825856"
}
W1_2：{
"dev_name": "Csmartt200",
"varible": {
"DCD4.DCSW_WD0_R0": "0.000000",
"DCD4.DCSW_WD0_R1": "429961.000000",
"DCD4.DCSW_WD0_R2": "0.000000",
"DCD4.DCSW_WD0_R3": "0.000000",
"DCD4.DCSW_WD0_R4": "0.000000",
"DCD4.DCSW_WD0_R5": "194956.000000",
"DCD4.DCSW_WD0_R6": "0.000000",
"DCD4.DCSW_WD0_R7": "3.000000",
"DCD4.DCSW_WD0_R8": "0.000000",
"DCD4.DCSW_WD0_R9": "0.000000",
"DCD4.DCSW_WD0_R10": "0.000000",
"DCD4.DCSW_WD0_R11": "0.000000",
"DCD4.DCSW_WD0_R12": "0.000000",
"DCD4.DCSW_WD0_R13": "1.000000",
"DCD4.DCSW_WD0_R14": "171524.000000"
},
"up_time": "20210301124825856"
}
————————————————————————————————問題
描述："DCD4.DCSW_WD0_R14": "171524.000000" 這是批次號遞增值，2個消費者收到的訊息同一批次
按照官方說法：使用直連交換機多個訂閱者就是輪詢機制，但是測驗發現不是，不知道如何解決？
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/264952.html
標籤：Java EE
下一篇：spring boot+swagger3 集成后,發現base url 老是 /,而所有的介面,卻都全部帶著 背景關系!請問這個應該如何處理!?