java - 有什么方法可以拆分通量流吗？

Question

我试图以多种方式分析数据（来自 RabbitMQ / spring cloud stream reactive）。我需要找到一种方法将测量通量分成多个“汇”。例如，我想做十秒钟的数据窗口，然后找到最大和最小的测量值。或者检查测量值是否在安全范围内，如果不在 - 打开警报（或发送电子邮件）。

我的项目： https ://github.com/Stiuil06/GreenRealTime/blob/master/grl-analysis/src/main/java/com/arturwegrzyn/grl/AnalysisApplication.java

我尝试了两种方法：

    @StreamListener
    public void receive1(@Input(AnalysisChannels.INPUT) Flux<String> measurements) {
        measurements
                .map(json -> gson.fromJson(json, Measurement.class))
                .filter(m -> m instanceof WaterLevel)
                .subscribe(m -> System.out.println(m));
    }

    @StreamListener
    public void receive2(@Input(AnalysisChannels.INPUT) Flux<String> measurements) {
        measurements
                .map(json -> gson.fromJson(json, Measurement.class))
                .filter(m -> m instanceof WaterLevel)
                .subscribe(m -> System.out.println(m));
    }

在这种情况下，一个事件只执行一个侦听器（随机第一次，第二次）

    @StreamListener
    public void receive2(@Input(AnalysisChannels.INPUT) Flux<String> measurements) {
        System.out.println("xyz");
        ConnectableFlux<String> publish = measurements.publish();
        publish
                .map(json -> gson.fromJson(json, Measurement.class))
                .filter(m -> m instanceof AirTemperature)
                .subscribe(m -> System.out.println(m));

        publish
                .map(json -> gson.fromJson(json, Measurement.class))
                .filter(m -> m instanceof AirTemperature)
                .subscribe(m -> System.out.println(m));
    }

在第二种情况下，我得到一个例外

2019-09-08 16:43:12.720 ERROR 12972 --- [nalysis-group-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.dataAnalysis'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[145], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=dataAnalysis, amqp_deliveryTag=1, deliveryAttempt=3, amqp_consumerQueue=dataAnalysis.realtime-analysis-group, amqp_redelivered=false, mqtt_receivedRetained=false, amqp_receivedRoutingKey=dataAnalysis, mqtt_duplicate=false, amqp_timestamp=Sun Sep 08 16:40:49 CEST 2019, amqp_messageId=e229ef37-4672-c524-e3bb-a04e607bb9cb, id=90ff1479-e363-e13e-ead7-aa7a64aaf612, amqp_consumerTag=amq.ctag-xcdhmSOud5ZYJquDKUgsiw, contentType=application/json, mqtt_receivedTopic=/measurement/si:mu:la:00/AirPressure, mqtt_receivedQos=1, timestamp=1567953789712}], failedMessage=GenericMessage [payload=byte[145], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=dataAnalysis, amqp_deliveryTag=1, deliveryAttempt=3, amqp_consumerQueue=dataAnalysis.realtime-analysis-group, amqp_redelivered=false, mqtt_receivedRetained=false, amqp_receivedRoutingKey=dataAnalysis, mqtt_duplicate=false, amqp_timestamp=Sun Sep 08 16:40:49 CEST 2019, amqp_messageId=e229ef37-4672-c524-e3bb-a04e607bb9cb, id=90ff1479-e363-e13e-ead7-aa7a64aaf612, amqp_consumerTag=amq.ctag-xcdhmSOud5ZYJquDKUgsiw, contentType=application/json, mqtt_receivedTopic=/measurement/si:mu:la:00/AirPressure, mqtt_receivedQos=1, timestamp=1567953789712}]
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:401)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:205)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.access$1200(AmqpInboundChannelAdapter.java:57)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.lambda$onMessage$0(AmqpInboundChannelAdapter.java:223)
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:180)
    at org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter$Listener.onMessage(AmqpInboundChannelAdapter.java:220)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1542)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1468)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1456)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1451)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1400)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:870)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:854)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:78)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1137)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1043)
    at java.base/java.lang.Thread.run(Thread.java:835)
Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[145], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=dataAnalysis, amqp_deliveryTag=1, deliveryAttempt=3, amqp_consumerQueue=dataAnalysis.realtime-analysis-group, amqp_redelivered=false, mqtt_receivedRetained=false, amqp_receivedRoutingKey=dataAnalysis, mqtt_duplicate=false, amqp_timestamp=Sun Sep 08 16:40:49 CEST 2019, amqp_messageId=e229ef37-4672-c524-e3bb-a04e607bb9cb, id=90ff1479-e363-e13e-ead7-aa7a64aaf612, amqp_consumerTag=amq.ctag-xcdhmSOud5ZYJquDKUgsiw, contentType=application/json, mqtt_receivedTopic=/measurement/si:mu:la:00/AirPressure, mqtt_receivedQos=1, timestamp=1567953789712}]
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:138)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:105)
    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:73)
    ... 23 more

通量/反应流拆分应该如何？或者也许我应该以不同的方式处理问题？

标签： javaspring-integrationfluxproject-reactorspring-cloud-stream

Answer 1

快速说明：我们将在 3.0 中一起弃用响应式模块，因为已经使用 spring-cloud-function 编程模型处理了相同的模块。这是您的代码的外观。请注意，不再需要 EnableBinding、StreamListener 和其他注释。只是@Bean，我们会

@Bean
public Consumer<Flux<String>> sink() {
    return measurements -> {
        ConnectableFlux<String> publish = measurements.publish();
        publish.connect();
        ...
    };
}

这里有更多 - https://spring.io/blog/2019/08/19/announcing-spring-cloud-stream-horsham-m3-3-0-0-m3