Kafka——Spring的云流

Question

我正在尝试用《Spring的云流》和《Kafka》。下面是示例代码。但它似乎没有任何作用。它总是创建一个名为“输出”的主题。但这些价值观尚未公布。

应用亚马尔

spring.cloud.stream:
  function:
    definition: streamSupplier
  bindings:
    streamSupplier-out-0:
      destination: numbers

我的目标就是创造价值。

@SpringBootApplication
@EnableBinding(Source.class)
public class CloudStreamDemoApplication {

    private AtomicInteger atomicInteger = new AtomicInteger();
    public static void main(String[] args) {
        SpringApplication.run(CloudStreamDemoApplication.class, args);
    }

    @Bean
    public Supplier<Integer> streamSupplier(){
        return () -> {
            System.out.println("Publishing : " + atomicInteger.incrementAndGet());
            return atomicInteger.get();
        };
    }
}

依赖性-2.2.6。释放

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-binder-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>

Answer 1

注释@EnableBinding导致了上述问题。

阅读以下spring Docs摘录：

与以前版本的spring cloud stream依赖@EnableBinding和@StreamListener注释不同，上面的示例看起来与任何普通的spring启动应用程序没有什么不同。它定义了一个类型为Function的bean，并且它是。那么，它是如何成为spring cloud stream应用程序的呢？它变成了spring cloud stream应用程序，只需在类路径上存在spring cloud stream和绑定器依赖项以及自动配置类，就可以有效地将引导应用程序的上下文设置为spring cloud stream应用程序。在此上下文中，类型为Supplier、Function或Consumer的bean被视为事实上的消息处理程序，触发绑定到所提供绑定器公开的目的地，遵循特定的命名约定和规则，以避免额外的配置

Answer 2

您需要从类中删除@EnableBinding（Source.class）。如果存在这种情况，功能绑定将不会发生。