我试着用Spring的云流与Kafka。下面是示例代码。但它似乎什么也没做。它总是创建一个名为“output”的主题。但这些数值并未公布。
应用程序.yaml
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.release
<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>
2条答案
按热度按时间ycl3bljg1#
注解@enablebinding导致了上述问题。
阅读以下spring文档摘录:
Unlike previous versions of spring-cloud-stream which relied on @EnableBinding and @StreamListener annotations, the above example looks no different then any vanilla spring-boot application. It defines a single bean of type Function and that it is. So, how does it became spring-cloud-stream application? It becomes spring-cloud-stream application simply based on the presence of spring-cloud-stream and binder dependencies and auto-configuration classes on the classpath effectively setting the context for your boot application as spring-cloud-stream application. And in this context beans of type Supplier, Function or Consumer are treated as defacto message handlers triggering binding of to destinations exposed by the provided binder following certain naming conventions and rules to avoid extra configuration.
vwhgwdsa2#
你需要移除
@EnableBinding(Source.class)
从班级里。如果存在这种情况,函数绑定将不会发生。