使用Spring在RabbitMQ中创建新队列

Question

我在 RabbitMQ 中创建新队列时遇到了一些问题。我只创建一个使用者客户端，该客户端将使用来自另一个微服务的消息。

这是我到目前为止所做的。

应用程序属性：

spring.rabbitmq.addresses=SCGLCCRAMQD0005.SCGER.DEV.CORP:5672
spring.rabbitmq.username=people-consumer
spring.rabbitmq.password=*************
spring.rabbitmq.vhost=PEOPLE

peopleevents.queue=qu-people-cores-update
peopleevents.exchange=ex-people-updates
peopleevents.routingkey=ONLINE.UPDATE.PERSONF.PERSONF

配置类：

@Configuration
public class MQConfig {


    @Value("${peopleevents.queue}")
    public String queue;

    @Value("${peopleevents.exchange}")
    public String exchange;

    @Value("${peopleevents.routingkey}")
    public String routingKey;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.addresses}")
    private String address;

    @Value("${spring.rabbitmq.vhost}")
    private String vHost;


    @Bean
    public Queue queue() {
        return new Queue(queue, true, false, false);
    }

    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(exchange);
    }

    //@Bean
    //Exchange myExchange() {
     //   return ExchangeBuilder.topicExchange(exchange).durable(true).build();
    //}

    @Bean
    public Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with(routingKey);
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public AmqpTemplate template(ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }

    @Bean
    public ConnectionFactory connectionFactory() throws IOException {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(address);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(vHost);
        connectionFactory.setPublisherReturns(true);

        return connectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }

}

和侦听器类：

@Service
public class RabbitMQService {

        private final Logger logger = LoggerFactory.getLogger(RabbitMQService.class);

        @RabbitListener(queues = "${peopleevents.queue}")
        public void receivedMessage(@Payload Message message) {
            logger.info("User Details Received is.. " + message.toString());
        }
    }

当我运行这个程序时，我有一条ACCESS_REFUSED消息，但我不知道为什么。我错过了什么吗？？

谢谢

Answer 1

您可以使用 RabbitAdmin 动态创建队列，因此您需要将 RabbitAdmin bean 添加到配置类中：

@Bean
public AmqpAdmin amqpAdmin() {
    return new RabbitAdmin(connectionFactory());
}

然后，您可以使用绑定和队列配置自动连线 RabbitAdmin：

@Autowired
private AmqpAdmin admin;

@Autowired
private Binding binding;

@Autowired
private Queue queue;

最后，您可以根据需要创建队列：

admin.declareQueue(queue);
admin.declareBinding(binding);

您可以在配置或服务类中创建队列。

您可以在构造函数中的服务类中实现它：

@Service
public class RabbitMQService {

        private final Logger logger = LoggerFactory.getLogger(RabbitMQService.class);

        private AmqpAdmin admin;

        private Binding binding;

        private Queue queue;

        @Autowired
        public RabbitMQService (AmqpAdmin admin, Binding binding, Queue queue) {
                this.admin = admin;
                this.binding = binding;
                this.queue = queue;
                admin.declareQueue(queue);
                admin.declareBinding(binding);
        }


        @RabbitListener(queues = "${peopleevents.queue}")
        public void receivedMessage(@Payload Message message) {
            logger.info("User Details Received is.. " + message.toString());
        }
}