如何在Spring AMQP中以编程方式将异步使用方订阅到队列？

debugcn 发表于 Dev

丹尼尔·切雷塞多（Daniel Cerecedo）

Spring AMQP文档仅显示如何在应用程序初始化时使用来以编程方式订阅队列~MessageListenerContainer。

现在，代码如下所示：

public void subscribe(EventType eventType, Object consumer) {
        Assert.notNull(eventType);
        Assert.notNull(eventType.toString());
        Assert.isTrue(!eventType.toString().isEmpty());

        Queue queue=new Queue("", false, true, true);
        Map<String, Exchange> beanMap=context.getBeansOfType(Exchange.class);
        if(beanMap!=null&&!beanMap.isEmpty()){
            Exchange exchange=null;
            boolean found=false;
            for(String key : beanMap.keySet()) {
                exchange=beanMap.get(key);
                if(getExchangeName(eventType.toString()).equals(exchange.getName())){
                    found=true;
                    break;
                }
            }
            if(found){
                amqpAdmin.declareQueue(queue);
                amqpAdmin.declareBinding(new Binding(queue.getName(), Binding.DestinationType.QUEUE, exchange.getName(), "", null));
                SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
                container.setConnectionFactory(connectionFactory);
                container.setQueueNames(queue.getName());
                container.setMessageListener(new MessageListenerAdapter(consumer));
            }else{
                //TODO
            }
        }
    }

由于SimpleMessageListenerContainer是生命周期Bean，因此除非在@ConfigurationBean中使用，否则它将无法正常工作

我需要在运行时以编程方式创建预订，以便Bean从队列中获取事件。任何想法如何做到这一点？

丹尼尔·切雷塞多（Daniel Cerecedo）

Gary Russells的回答是正确的，但是我的代码仍然存在一些剩余错误：

如果您创建的队列没有名称，RabbitMQ ~~创建时~~ 会给它一个随机的唯一名称。声明队列时必须检索它，然后在声明容器和绑定时使用此名称

为了清楚起见，我在最终版本中添加了修改后的代码块：

    public void subscribe(EventType eventType, AbstractEventConsumer consumer) {
    Assert.notNull(eventType);
    Assert.notNull(eventType.toString());
    Assert.isTrue(!eventType.toString().isEmpty());

    Queue queue=new Queue("", false, true, true);
    Map<String, Exchange> beanMap=context.getBeansOfType(Exchange.class);
    if(beanMap!=null&&!beanMap.isEmpty()){
        Exchange exchange=null;
        boolean found=false;
        for(String key : beanMap.keySet()) {
            exchange=beanMap.get(key);
            if(getExchangeName(eventType.toString()).equals(exchange.getName())){
                found=true;
                break;
            }
        }
        if(found){
            String queueName=amqpAdmin.declareQueue(queue);
            amqpAdmin.declareBinding(new Binding(queueName, Binding.DestinationType.QUEUE, exchange.getName(), "", null));
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            container.setQueueNames(queueName);
            container.setMessageListener(new MessageListenerAdapter(consumer));
            container.start();
        }else{
            //TODO
        }
    }
}