java - 使用 Spring Cloud Stream Source 将方法结果发送到流

Question

我正在尝试在 Spring Boot 应用程序中创建一个 Spring Cloud Stream Source Bean，它只是将方法的结果发送到流(底层 Kafka 主题绑定(bind)到流)。

我见过的大多数 Stream 示例都使用 @InboundChannelAdapter 注释来使用轮询器将数据发送到流。但我不想使用轮询器。我试过将轮询器设置为空数组，但另一个问题是在使用 @InboundChannelAdapter 时你无法使用任何方法参数。

我正在尝试做的事情的总体概念是从入站流中读取。进行一些异步处理，然后将结果发布到出站流。因此，使用处理器似乎也不是一种选择。我正在将 @StreamListener 与 Sink channel 一起使用来读取入站流，这很有效。

这是我一直在尝试的一些代码，但这根本不起作用。我希望它会这么简单，因为我的水槽是但也许不是。寻找某人指出一个不是处理器的源示例(即不需要监听入站 channel )并且不使用 @InboundChannelAdapter 或给我一些设计提示以不同的方式完成我需要做的事情。谢谢!

@EnableBinding(Source.class)
public class JobForwarder {

   @ServiceActivator(outputChannel = Source.OUTPUT)
   @SendTo(Source.OUTPUT)
   public String forwardJob(String message) {
       log.info(String.format("Forwarding a job message [%s] to queue [%s]", message, Source.OUTPUT));
       return message;
   }
}

Answer 1

您的原始需求可以通过以下步骤实现。

1. 创建您的自定义绑定(bind)接口(interface)(您也可以使用默认的 @EnableBinding(Source.class))

   public interface CustomSource {
       String OUTPUT = "customoutput";
   
       @Output(CustomSource.OUTPUT)
       MessageChannel output();
   }
   

2. 注入(inject)你的绑定(bind) channel

   @Component
   @EnableBinding(CustomSource.class)
   public class CustomOutputEventSource {
   
       @Autowired
       private CustomSource customSource;
   
       public void sendMessage(String message) {
           customSource.output().send(MessageBuilder.withPayload(message).build());
       }
   }
   

3. 测试一下

   @RunWith(SpringRunner.class)
   @SpringBootTest
   public class CustomOutputEventSourceTest {
   
       @Autowired
       CustomOutputEventSource output;
   
       @Test
       public void sendMessage() {
           output.sendMessage("Test message from JUnit test");
       }
   }