如何通过多个 StreamListener Spring Cloud Stream 和 Kafka 流收听多个主题

Question

我想听下面的代码中的两个 Kafka 主题，有两个源事件需要处理并转换为另一个事件。

所以我想一口气听这两个事件EnableBinding

    @EnableBinding(PartnerOrderCancelledEventImporter.Targets.class)
    public class PartnerOrderCancelledEventImporter {
       @StreamListener(Targets.INPUT_ORDER_CANCELLED)
        @SendTo(Targets.OUTPUT)
        public KStream<?, TriggeringEvent> processOrderCancelled(KStream<?, OrderCancelledV1> input) {
            LogInfo("PartnerOrderCancelled-OrderCancelledV1 stream started");
            LogInfo("  KafkaBrokers: " + KafkaBrokers);
            return input
                    .filter((key, value) -> IsFFFaultAndNoRoutes(value))
                    .peek((key, value) -> LogInfo("OrderCancelle", new LogObject(value)))
                    .map((key, value) -> KeyValue.pair(key, new TriggeringEvent(value)));
        }
        @StreamListener(Targets.INPUT_ORDER_ITEM_STOCK_CHECKED)
        @SendTo(Targets.OUTPUT)
        public KStream<?, TriggeringEvent> processOrderItemStockChecked(KStream<?, OrderItemStockCheckedV1> input) {
            LogInfo("PartnerOrderCancelled-OrderItemStockCheckedV1 stream started");
            LogInfo("  KafkaBrokers: " + KafkaBrokers);
            return input
                    .filter((key, value) -> IsItemNoStock(value))
                    .peek((key, value) -> LogInfo("OrderItemStockChecke", new LogObject(value)))
                    .map((key, value) -> KeyValue.pair(key, new TriggeringEvent(value)));
        }

        public interface Targets {
            String INPUT_ORDER_CANCELLED = "partnerOrderCancelledInputOrderCancelled";
            String INPUT_ORDER_ITEM_STOCK_CHECKED = "partnerOrderCancelledInputOrderItemStockChecked";
            String OUTPUT = "triggeringEventsOutputPartnerOrderCancelled";
            @Input(INPUT_ORDER_CANCELLED)
            KStream<?, ?> inputOrderCancelled();
            @Input(INPUT_ORDER_ITEM_STOCK_CHECKED)
            KStream<?, ?> inputOrderItemStockChecked();
            @Output(OUTPUT)
            KStream<?, ?> output();
        }
   spring.cloud.stream.bindings.triggeringEventsOutputPartnerOrderCancelled.destination=dev.comms.triggeringevents.TriggeringEvent-events-1.0
    spring.cloud.stream.bindings.partnerOrderCancelledInputOrderItemStockChecked.destination=dev.ecom.order-management-service.order-item-stock-checked-events-v1
    spring.cloud.stream.bindings.partnerOrderCancelledInputOrderItemStockChecked.group=TriggeringEvents-PartnerOrderCancelled
    spring.cloud.stream.bindings.partnerOrderCancelledInputOrderCancelled.destination=dev.ecom.order-management-service.order-cancelled-events-v1
    spring.cloud.stream.bindings.partnerOrderCancelledInputOrderCancelled.group=TriggeringEvents-PartnerOrderCancelled

抛出在异常下方

org.springframework.beans.factory.BeanInitializationException: Cannot setup StreamListener for public org.apache.kafka.streams.kstream.KStream com.farfetch.communication.triggeringeventsimporter.PartnerOrderCancelledEventImporter.processOrderItemStockChecked(org.apache.kafka.streams.kstream.KStream); nested exception is java.lang.reflect.UndeclaredThrowableException
    at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:195)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.doPostProcess(StreamListenerAnnotationBeanPostProcessor.java:195)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.lambda$postProcessAfterInitialization$0(StreamListenerAnnotationBeanPostProcessor.java:167)
    at java.lang.Iterable.forEach(Iterable.java:75)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.injectAndPostProcessDependencies(StreamListenerAnnotationBeanPostProcessor.java:285)
    at org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor.afterSingletonsInstantiated(StreamListenerAnnotationBeanPostProcessor.java:105)
    。。。。。。。。。。。。。。。。。。。。。。。。。。。。。
    at org.springframework.cloud.stream.binder.kafka.streams.KStreamStreamListenerResultAdapter.adapt(KStreamStreamListenerResultAdapter.java:41)
    at org.springframework.cloud.stream.binder.kafka.streams.KStreamStreamListenerResultAdapter.adapt(KStreamStreamListenerResultAdapter.java:31)
    at org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsStreamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(KafkaStreamsStreamListenerSetupMethodOrchestrator.java:187)
    ... 39 more
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.cloud.stream.binder.kafka.streams.KStreamBoundElementFactory$KStreamWrapperHandler.invoke(KStreamBoundElementFactory.java:99)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:185)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
    ... 43 more
Caused by: java.lang.IllegalArgumentException: delegate already set to org.apache.kafka.streams.kstream.internals.KStreamImpl@491cafec

Answer 1

您收到此异常是因为您尝试从两个处理器(输出(绑定相同的出站 KStream。是否有可能为第二个处理器添加另一个输出绑定？这应该解决您遇到的这个特定错误。另一方面，如果这是您的用例要求的要求，那么目前活页夹不支持该要求(不过这可能是一个不错的功能(。作为一种解决方法，您可以使第二个处理器不返回任何内容，然后在KStream上调用 to 方法以手动发送它。