与Kafka Spring Cloud Stream多个@EnableBinding [英] Multiple @EnableBinding with Kafka Spring Cloud Stream

问题描述

我正在尝试设置一个侦听Kafka的Spring Boot应用程序.

I'm trying to set an Spring Boot Application listening to Kafka.

我正在使用Kafka Streams Binder.

I'm using Kafka Streams Binder.

一个简单的@EnableBinding

@EnableBinding(StreamExample.StreamProcessor.class)
public class StreamExample {

    @StreamListener(StreamProcessor.INPUT)
    @SendTo(StreamProcessor.OUTPUT)
    public KStream<String, String> process(KStream<String, String> input) {

        logger.info("Stream listening");

        return input
                .peek(((key, value) -> logger.info("key = {} value = {}", key, value)));
    }

    interface StreamProcessor {

        String INPUT = "input_1";
        String OUTPUT = "output_1";

        @Input(INPUT)
        KStream<String, String> input();

        @Output(OUTPUT)
        KStream<String, String> outputProcessed();
    }
}

和application.yml

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            brokers: localhost:29092
      bindings:
         input_1:
           destination: mytopic1
           group: readgroup
         output_1:
           destination: mytopic2
         input_2:
           destination: mytopic3
           group: readgroup
         output_2:
           destination: mytopic4
  application:
    name: stream_s1000_app

一切正常.

但是，如果我尝试添加具有其他绑定的第二个类，则会发生以下错误:

But if I try to add a second class with other binding, the following error occurs:

以下订阅的主题未分配给任何成员:[mytopic1]

第二个绑定的示例:

@EnableBinding(StreamExampleBindingTwo.StreamProcessor.class)
public class StreamExampleBindingTwo {

    @StreamListener(StreamProcessor.INPUT)
    @SendTo(StreamProcessor.OUTPUT)
    public KStream<String, String> process(KStream<String, String> input) {

        logger.info("Stream listening binding two");

        return input
                .peek(((key, value) -> logger.info("key = {} value = {}", key, value)));
    }

    interface StreamProcessor {

        String INPUT = "input_2";
        String OUTPUT = "output_2";

        @Input(INPUT)
        KStream<String, String> input();

        @Output(OUTPUT)
        KStream<String, String> outputProcessed();
    }
}

我想念的是什么?我不能在同一应用程序中使用多个输入主题和多个输出吗?是否有与application.name相关的内容?

What I'm missing? Can't I use multiple input topics and multiple output in the same application? There is something related to application.name?

推荐答案

我刚刚尝试了一个应用程序，并且该方法有效.当同一应用程序中有多个处理器时，需要确保每个处理器都具有自己的应用程序ID. 参见下面的application.yml中的两个输入，我如何有2个不同的应用程序ID.

I just tried an app and that worked. When you have multiple processors in the same application, you need to make sure that each processor gets its own application id. See below how I have 2 distinct application id's for both inputs in the application.yml.

我看到两个处理器都已登录到控制台.此外，还看到了有关输出主题的消息.

I saw both processors are getting logged on the console. Also, saw the messages on the output topics.

@SpringBootApplication
@EnableBinding({So54522918Application.StreamProcessor1.class, So54522918Application.StreamProcessor2.class})
public class So54522918Application {

    public static void main(String[] args) {
        SpringApplication.run(So54522918Application.class, args);
    }

    @StreamListener(StreamProcessor1.INPUT)
    @SendTo(StreamProcessor1.OUTPUT)
    public KStream<String, String> process1(KStream<String, String> input) {

        System.out.println("Stream listening");

        return input
                .peek(((key, value) -> System.out.println("key = " + key +", value = " + value)));
    }

    @StreamListener(StreamProcessor2.INPUT)
    @SendTo(StreamProcessor2.OUTPUT)
    public KStream<String, String> process2(KStream<String, String> input) {

        System.out.println("Stream listening binding two");

        return input
                .peek(((key, value) -> System.out.println("key = " + key +", value = " + value)));
    }

    interface StreamProcessor1 {

        String INPUT = "input_1";
        String OUTPUT = "output_1";

        @Input(INPUT)
        KStream<String, String> input();

        @Output(OUTPUT)
        KStream<String, String> outputProcessed();
    }

    interface StreamProcessor2 {

        String INPUT = "input_2";
        String OUTPUT = "output_2";

        @Input(INPUT)
        KStream<String, String> input();

        @Output(OUTPUT)
        KStream<String, String> outputProcessed();
    }

}

application.yml的相关部分

Relevant part of application.yml

spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms: 1000
spring.cloud.stream.kafka.streams:
  binder.configuration:
    default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
  bindings.input_1.consumer.application-id: process-1
  bindings.input_2.consumer.application-id: process-2
spring.cloud.stream.bindings.input_1:
  destination: mytopic1
spring.cloud.stream.bindings.output_1:
  destination: mytopic2
spring.cloud.stream.bindings.input_2:
  destination: mytopic3
spring.cloud.stream.bindings.output_2:
  destination: mytopic4

这篇关于与Kafka Spring Cloud Stream多个@EnableBinding的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！