如何使用不推荐使用 @Output 的 Spring Cloud Stream 仅生成一条消息，或关闭轮询？

Question

我正在尝试使用没有任何已弃用的类/方法或注释的 Spring Cloud 向 Kafka 发布 一条 消息。我还希望能够轻松更改有效负载。

所以为了清楚起见，我尽量不使用 deprecated @Output 注释，也不要使用任何 KafkaTemplate。

我的配置：

spring:
  cloud:
    stream:
      bindings:
        message-out-0:
          destination: ${spring.application.name}
          producer:
            key:
              serializer:
                type: string
                format: utf-8
                charset: utf-8
            value:
              serializer:
                type: string
                format: utf-8
                charset: utf-8

我的代码 - 到目前为止我尝试过的：

@Component
@RequiredArgsConstructor
public class ApplicationAnnouncer implements CommandLineRunner {
    
    private final MessageService messageService;
    
    @Override
    public void run(String... args) throws Exception {
        messageService.value = "Application started...";
        messageService.message();
    }
}

一次尝试：

@Configuration
public class MessageService {
     public Object value;
     
     @Bean
     public Supplier<Message<?>> message () {
          return () -> MessageBuilder.withPayload(value).build();
     }
}

另一个尝试：

@Configuration
public class MessageService {
     public Object value;
     
     @Bean
     public Supplier<Flux<?>> message () {
          return () -> Flux.fromStream(Stream.generate(() -> {
               try {
                    Thread.sleep(1000);
                    return value;
               } catch (Exception e) {
                    // ignore
               }
               return null;
          })).subscribeOn(Schedulers.elastic()).share();
     }
}

两次尝试的控制台使用者输出：

Hello World!
Hello World!
Hello World!
Hello World! // ... Repeated every second

documentation 声明：

框架提供了一个默认的轮询机制（回答“谁？”的问题），它将触发供应商的调用，默认情况下它会每秒执行一次（回答“多久一次？”的问题）。

但如果我不希望它每秒轮询一次呢？

我向 MessageService 提供消息的方式很奇怪...... 是配置吗？还是服务？

我还没有找到将 ONE CUSTOMIZABLE MESSAGE 推送到 Kafka 的最基本示例。

Answer 1

您可以使用StreamBridge 访问云流绑定：

@Component
@RequiredArgsConstructor
public class ApplicationAnnouncer implements CommandLineRunner {

    private final StreamBridge streamBridge;
    
    @Override
    public void run(String... args) throws Exception {
        streamBridge.send("message-out-0", "Application started...");
    }
}

第一个字符串是应用程序设置中提供的绑定名称，从提供函数的 bean 派生。

您甚至不需要派生 binding-name 的实际 bean。在这种情况下，任何名称都可以。

---

您可以找到一些示例here。