springBoot整合rabbitMQ，动态创建队列，动态监听。（三）

青旅半醒 2022-12-21 04:43 152阅读 0赞

一，

动态监听和队列的使用场景主要是项目不确定队列的名称等信息，根据业务队列变化，或者在大量数据处理的创景中一个或者几个队列处理不过来，需要随着数据增加消费者队列也要动态增长

下面是简单使用：

SimpleMessageListenerContainer类是rabbitMQ中专门负责监听队列，维护队列的添加，移除等；

MessageListener是rabbitMQ中专门负责接收消息以后处理消息的接口，SimpleMessageListenerContainer在监听到消息以后会调用MessageListener，将消息进行处理；

rabbitMQ的动态监听主要靠这两个类就可以做一些简单的实现

二，

自己实现消息处理的接口MessageListener

@Service
    public class TailWindServiceImpl implements MessageListener {
    
       
    
        @Override
        public void onMessage(Message message) {
    
            byte[] body = message.getBody();
            String s = new String(body);
             System.out.println("接收到消息");
        }
    }

三，

将处理类放入到消息监听里面

@Configuration
    public class RabbitMQConfig {
    
        @Autowired
        private TailWindServiceImpl tailWindService;
    
    
      @Bean(name = "primarySimpleMessageListenerContainer")
        public SimpleMessageListenerContainer primaryMessageListenerContainer(@Qualifier("primaryConnectionFactory") ConnectionFactory connectionFactory) {
            SimpleMessageListenerContainer container = new         
            SimpleMessageListenerContainer(connectionFactory);
            container.setDefaultRequeueRejected(false);
            container.setAcknowledgeMode(AcknowledgeMode.AUTO);
            container.setMessageListener(tailWindService);
            container.start();
            return container;
        }
    
    }

四，

动态的创建队列，将队列放入监听器

@Autowired
        @Qualifier("primaryRabbitAdmin")
        private RabbitAdmin rabbitAdmin;
        @Autowired
        @Qualifier("primarySimpleMessageListenerContainer")
        private SimpleMessageListenerContainer messageListenerContainer;
    
    
    
    
                 //创建队列，绑定交换机
    
                Queue queue = new Queue(invitationCode);
                FanoutExchange exchange = new 
                FanoutExchange(RabbitMQConfiguration.TERMINAL_FANOUT_EXCHANGE_cloud);
                rabbitAdmin.declareQueue(queue);
                rabbitAdmin.declareExchange(exchange);
                rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange));
                messageListenerContainer.addQueueNames(invitationCode);
                messageListenerContainer.start();