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

通量/ react 流拆分应该是什么样的？或者也许我应该以不同的方式处理问题？

Answer 1

我启动了第二种情况下显示的代码。
那里缺少 connect()。 它应该看起来像:

    public void receive2(@Input(AnalysisChannels.INPUT) Flux<String> measurements) {
        ConnectableFlux<String> publish = measurements.publish();
        publish.connect();
        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));
    }

无论如何，如果有人对我的逻辑有更好的解决方案，请随时写出来!
我将感谢你的帮助。