首页 文章

Spring Amqp：将SimpleRoutingConnectionFactory与@RabbitListener混合使用

提问于
浏览 次
0

我有一个应用程序，它将侦听多个队列，这些队列在不同的vhost上声明 . 我使用SimpleRoutingConnectionFactory来存储connectionFactoryMap，我希望用@RabbitListener设置我的监听器 .

根据Spring AMQP doc：

从1.4版开始，您可以在SimpleMessageListenerContainer中配置路由连接工厂 . 在这种情况下，队列名称列表用作查找键 . 例如，如果使用setQueueNames（“foo，bar”）配置容器，则查找键将为“[foo，bar]”（无空格） .

我用 @RabbitListener(queues = "some-key") . 不幸的是， Spring 天抱怨"lookup key [null]" . 见下文 .

18：52：44.528 WARN --- [cTaskExecutor-1] osarlSimpleMessageListenerContainer：消费者引发异常，如果连接工厂支持它，处理可以重启java.lang.IllegalStateException：无法在org中为查找键[null]确定目标ConnectionFactory . springframework.amqp.rabbit.connection.AbstractRoutingConnectionFactory.determineTargetConnectionFactory（AbstractRoutingConnectionFactory.java:119）位于org.springframework.amqp.rabbit.connection的org.springframework.amqp.rabbit.connection.AbstractRoutingConnectionFactory.createConnection（AbstractRoutingConnectionFactory.java:97） . 在org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.getTransactionalResourceHolder（ConnectionFactoryUtils.java）的org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.doGetTransactionalResourceHolder（ConnectionFactoryUtils.java:140）的ConnectionFactoryUtils $ 1.createConnection（ConnectionFactoryUtils.java:90） ：76）在org.springframework.amqp.rabbit.listener.Blocki ngQueueConsumer.start（BlockingQueueConsumer.java:472）atg.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer $ AsyncMessageProcessingConsumer.run（SimpleMessageListenerContainer.java:1306）at java.lang.Thread.run（Thread.java:745）

  • 我做错了吗？如果将queues属性用作查找键（用于连接工厂查找），我应该使用什么来指定我想要侦听的队列？

  • 最终，我希望能够进行程序化/动态监听器设置 . 如果我使用“程序化 endpoints 注册”，我是否应该删除“注释驱动的侦听器 endpoints ”？我喜欢“注释驱动的侦听器 endpoints ”，因为侦听器可以有多个带有不同传入数据类型的消息句柄作为参数，这非常干净整洁 . 如果我使用Programmatic Endpoint Registration，我将不得不解析Message输入变量，并根据消息类型/内容调用我的特定自定义消息处理程序 .

EDIT: 嗨Gary，我稍微修改了你的代码＃2，以便它使用Jackson2JsonMessageConverter来序列化类对象（在RabbitTemplate bean中），并使用它将它们反序列化回对象（在inboundAdapter中） . 我还删除了@RabbitListener，因为在我的情况下，所有的监听器都会在运行时添加 . 现在fooBean可以接收整数，字符串和TestData消息而没有任何问题！留下的唯一问题是该计划不断报告警告：

“[erContainer＃0-1] o.s.a.r.l.SimpleMessageListenerContainer：消费者引发异常，如果连接工厂支持，处理可以重启

java.lang.IllegalStateException：无法确定查找键[null]的目标ConnectionFactory“ . 对于完整的堆栈跟踪，请参见底部 .

我错过了什么吗？

@SpringBootApplication
public class App2 implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(App2.class, args);
    }

    @Autowired
    private IntegrationFlowContext flowContext;

    @Autowired
    private ConnectionFactory routingCf;

    @Autowired
    private RabbitTemplate template;

    @Override
    public void run(String... args) throws Exception {
        // dynamically add a listener for queue qux
        IntegrationFlow flow = IntegrationFlows.from(Amqp.inboundAdapter(this.routingCf, "qux").messageConverter(new Jackson2JsonMessageConverter()))
                .handle(fooBean())
                .get();
        this.flowContext.registration(flow).register();

        // now test it
        SimpleResourceHolder.bind(this.routingCf, "[qux]");
        this.template.convertAndSend("qux", 42);
        this.template.convertAndSend("qux", "fizbuz");
        this.template.convertAndSend("qux", new TestData(1, "test"));
        SimpleResourceHolder.unbind(this.routingCf);
    }

    @Bean
    RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(routingCf);
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        return template;
    }

    @Bean
    @Primary
    public ConnectionFactory routingCf() {
        SimpleRoutingConnectionFactory rcf = new SimpleRoutingConnectionFactory();
        Map<Object, ConnectionFactory> map = new HashMap<>();
        map.put("[foo,bar]", routedCf());
        map.put("[baz]", routedCf());
        map.put("[qux]", routedCf());
        rcf.setTargetConnectionFactories(map);
        return rcf;
    }

    @Bean
    public ConnectionFactory routedCf() {
        return new CachingConnectionFactory("127.0.0.1");
    }

    @Bean
    public Foo fooBean() {
        return new Foo();
    }

    public static class Foo {

        @ServiceActivator
        public void handleInteger(Integer in) {
            System.out.println("int: " + in);
        }

        @ServiceActivator
        public void handleString(String in) {
            System.out.println("str: " + in);
        }

        @ServiceActivator
        public void handleData(TestData data) {
            System.out.println("TestData: " + data);
        }
    }
}

完整堆栈跟踪：

2017-03-15 21:43:06.413  INFO 1003 --- [           main] hello.App2                               : Started App2 in 3.003 seconds (JVM running for 3.69)
2017-03-15 21:43:11.415  WARN 1003 --- [erContainer#0-1] o.s.a.r.l.SimpleMessageListenerContainer : Consumer raised exception, processing can restart if the connection factory supports it

java.lang.IllegalStateException: Cannot determine target ConnectionFactory for lookup key [null]
    at org.springframework.amqp.rabbit.connection.AbstractRoutingConnectionFactory.determineTargetConnectionFactory(AbstractRoutingConnectionFactory.java:119) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.AbstractRoutingConnectionFactory.createConnection(AbstractRoutingConnectionFactory.java:97) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:1430) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1411) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1387) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.core.RabbitAdmin.initialize(RabbitAdmin.java:500) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.core.RabbitAdmin$11.onCreate(RabbitAdmin.java:419) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.CompositeConnectionListener.onCreate(CompositeConnectionListener.java:33) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createConnection(CachingConnectionFactory.java:571) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils$1.createConnection(ConnectionFactoryUtils.java:90) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.doGetTransactionalResourceHolder(ConnectionFactoryUtils.java:140) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactoryUtils.java:76) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.start(BlockingQueueConsumer.java:505) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1382) ~[spring-rabbit-1.7.1.RELEASE.jar:na]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_112]
rabbitmq spring-amqp

1 回答

  • 0

    请显示您的配置 - 它对我来说很好...

    @SpringBootApplication
public class So42784471Application {

    public static void main(String[] args) {
        SpringApplication.run(So42784471Application.class, args);
    }

    @Bean
    @Primary
    public ConnectionFactory routing() {
        SimpleRoutingConnectionFactory rcf = new SimpleRoutingConnectionFactory();
        Map<Object, ConnectionFactory> map = new HashMap<>();
        map.put("[foo,bar]", routedCf());
        map.put("[baz]", routedCf());
        rcf.setTargetConnectionFactories(map);
        return rcf;
    }

    @Bean
    public ConnectionFactory routedCf() {
        return new CachingConnectionFactory("10.0.0.3");
    }

    @RabbitListener(queues = { "foo" , "bar" })
    public void foobar(String in) {
        System.out.println(in);
    }

    @RabbitListener(queues = "baz")
    public void bazzer(String in) {
        System.out.println(in);
    }

}

    关于你的第二个问题，你可以手动构建 endpoints ，但它可能更容易在Spring Integration _2523514中使用类似的功能 .

    我将很快更新这个答案的详细信息 .

    EDIT

    这是使用Spring Integration技术在运行时动态添加多方法侦听器的更新...

    @SpringBootApplication
public class So42784471Application implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(So42784471Application.class, args);
    }

    @Autowired
    private IntegrationFlowContext flowContext;

    @Autowired
    private ConnectionFactory routingCf;

    @Autowired
    private RabbitTemplate template;

    @Override
    public void run(String... args) throws Exception {
        // dynamically add a listener for queue qux
        IntegrationFlow flow = IntegrationFlows.from(Amqp.inboundAdapter(this.routingCf, "qux"))
                .handle(fooBean())
                .get();
        this.flowContext.registration(flow).register();

        // now test it
        SimpleResourceHolder.bind(this.routingCf, "[qux]");
        this.template.convertAndSend("qux", 42);
        this.template.convertAndSend("qux", "fizbuz");
        SimpleResourceHolder.unbind(this.routingCf);
    }


    @Bean
    @Primary
    public ConnectionFactory routingCf() {
        SimpleRoutingConnectionFactory rcf = new SimpleRoutingConnectionFactory();
        Map<Object, ConnectionFactory> map = new HashMap<>();
        map.put("[foo,bar]", routedCf());
        map.put("[baz]", routedCf());
        map.put("[qux]", routedCf());
        rcf.setTargetConnectionFactories(map);
        return rcf;
    }

    @Bean
    public ConnectionFactory routedCf() {
        return new CachingConnectionFactory("10.0.0.3");
    }

    @RabbitListener(queues = { "foo" , "bar" })
    public void foobar(String in) {
        System.out.println(in);
    }

    @RabbitListener(queues = "baz")
    public void bazzer(String in) {
        System.out.println(in);
    }

    @Bean
    public Foo fooBean() {
        return new Foo();
    }

    public static class Foo {

        @ServiceActivator
        public void handleInteger(Integer in) {
            System.out.println("int: " + in);
        }

        @ServiceActivator
        public void handleString(String in) {
            System.out.println("str: " + in);
        }

    }

}
    回复于

相关问题