java - 是否可以使一种方法(处理程序)从 Spring Cloud Stream 中的多个入站 channel 接收消息？

Question

假设我对 Sink 接口(interface)有这个定义

public interface Sink {
  @Input("input")
  SubscribableChannel input();

  @Input("anotherInput")
  SubscribableChannel anotherInput();
}

并且以下 Controller 绑定(bind)到 RabbitMQ(所有设置均根据 Spring Cloud Stream Binder Rabbit 正确设置)

@Controller
@EnableBinding(Sink.class)
public class InputMessageController {

    @StreamListener("input")
    public void handle(String message) {
        System.out.println("Received from input: " + message);
    }

    @StreamListener("anotherInput")
    public void handleOther(String message) {
        System.out.println("Received from another input: " + message);
    }

虽然它可以工作，但由于重复等原因，它是糟糕的代码。

我想在这个 Controller 中设置一个处理程序来监听，换句话说，订阅 Sink 接口(interface)中的两个 channel ，并根据某些条件处理它们。到目前为止，我无法用标准的 @StreamListener 注释来解决这个问题，因为它显然不接受字符串数组( channel 名称)

我的目标是这样的:

@Controller
@EnableBinding(Sink.class)
public class InputMessageController {

    @StreamListener("input", "anotherInput")
    public void handle(String message) {
        System.out.println("Received " + message + "from " +
                   ((/*some condition*/) ? "1st" : "2nd") + " input");
    }

如果有任何解决问题的想法或有关 Spring Cloud Stream API 的解释，我将不胜感激!

Answer 1

用 Rabbitlistener 注释你的类并将你的队列声明为参数，然后用 Rabbithandler 注释一个方法。此方法接收所有已声明队列的输入。

@Controller
@EnableBinding(Sink.class)
@RabbitListener(queues = {"input", "anotherInput"})
public class InputMessageController {

    @RabbitHandler
    public void handle(String message) {
        System.out.println("Received " + message + "from " +
                   ((/*some condition*/) ? "1st" : "2nd") + " input");
    }