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

Question

我试图在Kafka上发送消息，当在特定表记录中插入时。我认为这部分应用程序将被视为供应商。

我有以下代码。

@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. 如果我声明一个“函数”，那么输入似乎将从主题接收并转发。我没有收到来自主题的数据。我正在从RESTAPI数据库接收数据

我不确定以下几点是否能有所帮助：但只要做一次基本的尝试，我就不会成功。因此，举个例子会很有帮助。

@Autowired
private StreamBridge streamBridge;

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

供参考：

• https://cloud.spring.io/spring-cloud-stream/reference/html/spring-cloud-stream.html#spring_cloud_function

Answer 1

跟踪奏效了

声明字段

  private final StreamBridge streamBridge;

声明的函数

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

调用这些函数时，会将消息放在主题上