我們可以批量使用彈簧集成在蚊子10消息加載組

Question

這是我如何定義我的mqtt連接使用春天integration.i我不知道這是否可能bt我們可以設置一個mqtt訂戶後得到10消息的加載。現在訂閱者在發佈消息之後就工作了。

@Autowired 
    ConnectorConfig config; 


    @Bean 
    public MqttPahoClientFactory mqttClientFactory() { 
     DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); 
     factory.setServerURIs(config.getUrl()); 
     factory.setUserName(config.getUser()); 
     factory.setPassword(config.getPass()); 
     return factory; 
    } 

    @Bean 
    public MessageProducer inbound() { 
     MqttPahoMessageDrivenChannelAdapter adapter = 
       new MqttPahoMessageDrivenChannelAdapter(config.getClientid(), mqttClientFactory(), "ALERT", "READING"); 

     adapter.setCompletionTimeout(5000); 
     adapter.setConverter(new DefaultPahoMessageConverter()); 
     adapter.setQos(1); 
     adapter.setOutputChannel(mqttRouterChannel()); 
     return adapter; 
    } 

    /**this is router**/ 
    @MessageEndpoint 
    public class MessageRouter { 

    private final Logger logger = LoggerFactory.getLogger(MessageRouter.class); 


    static final String ALERT = "ALERT"; 
    static final String READING = "READING"; 

    @Router(inputChannel = "mqttRouterChannel") 
    public String route(@Header("mqtt_topic") String topic){ 
     String route = null; 
     switch (topic){ 
      case ALERT: 
       logger.info("alert message received"); 
       route = "alertTransformerChannel"; 
       break; 
      case READING: 
       logger.info("reading message received"); 
       route = "readingTransformerChannel"; 
       break; 
     } 
     return route; 
    } 
} 

Answer 1

我需要批量達10條消息在同一時間

這不是一個MqttPahoMessageDrivenChannelAdapter責任。由泛美衛生組織客戶的性質

* @param topic name of the topic on the message was published to 
* @param message the actual message. 
* @throws Exception if a terminal error has occurred, and the client should be 
* shut down. 
*/ 
public void messageArrived(String topic, MqttMessage message) throws Exception; 

所以，我們不能批他們有這個通道適配器：

我們有MqttCallback使用這個語義。

我們可以從彈簧集成的角度來建議您的EIP實施。

在你的情況，你應該爲AggregatorFactoryBean@Bean添加@ServiceActivator是mqttRouterChannel之前，發送到路由器之前。

，也許就這麼簡單：

@Bean 
@ServiceActivator(inputChannel = "mqttAggregatorChannel") 
AggregatorFactoryBean mqttAggregator() { 
    AggregatorFactoryBean aggregator = new AggregatorFactoryBean(); 
    aggregator.setProcessorBean(new DefaultAggregatingMessageGroupProcessor()); 
    aggregator.setCorrelationStrategy(m -> 1); 
    aggregator.setReleaseStrategy(new MessageCountReleaseStrategy(10)); 
    aggregator.setExpireGroupsUponCompletion(true); 
    aggregator.setSendPartialResultOnExpiry(true); 
    aggregator.setGroupTimeoutExpression(new ValueExpression<>(1000)); 
    aggregator.setOutputChannelName("mqttRouterChannel"); 
    return aggregator; 
} 

看到Reference Manual的更多信息。