如何与Spring Cloud Streams供应商发送消息/数据？

Question

当特定表中存在插入时，我尝试在

kafka

上发送消息

Record

。 我认为申请的这一部分将被视为

Supplier

/

Producer

。

我有以下代码。

@Bean
public Supplier<RecordAlteredEvent> affectedRecordEventEmitter() {
    return (/*how do I pass the data?*/) -> {
            
        log.info("SENDING_MESSAGE TO RECORD_EVENT_TOPIC");
        return new RecordAlteredEvent();
    };
}

我其实想发送数据。所以，我正在寻找更像以下的东西：

@Bean
public Function<RecordAlteredEvent, RecordAlteredEvent> alteredRecordEventEmitter() {
    return (RecordAlteredEvent recordAlteredEvent) -> {
        log.info(SENDING_MESSAGE, VDP_USERS_EVENT_TOPIC, recordAlteredEvent.toString());
        return recordAlteredEvent;
    };
    }

这可能吗？我该如何配置呢？所以，例如，如果

1. 我声明了一个“供应商”，那么我如何向它传递数据？
2. 如果我声明一个“函数”，似乎输入将从

   topic

   接收并转发。我没有收到来自

   topic

   的数据。我正在从

   REST API

   /

   DB

   接收数据。

我不确定以下内容是否能有所帮助：但通过基本的尝试我无法成功。因此，一个例子会有很大帮助。

@Autowired
private StreamBridge streamBridge;

// Can following work somehow in-side a function that is, for example, in spring-boot-service  
streamBridge.send("myDestination", body);

供参考：

1. https://cloud.spring.io/spring-cloud-stream/reference/html/spring-cloud-stream.html#spring_cloud_function
2. https://tanzu.vmware.com/developer/guides/scs-gs/
   • 部分：生成贷款申请
3. 如何使用 Spring Cloud Kafka Stream 3.1 创建生产者

Answer 1

以下工作

声明字段

  private final StreamBridge streamBridge;

声明的函数

  public void delegateRecordsEventSupplier(@NotNull final RecordsEvent alteredRecordsEvent) {
    streamBridge.send("affectedRecordsEvent-out-0", alteredRecordsEvent);
  }

调用这些函数时会将消息放置在

topic

Answer 2

你可以使用各种方法来实现目标。 参考这个链接：

https://docs.spring.io/spring-cloud-stream/reference/4.1/spring-cloud-stream/生产和消费-messages.html