rabbitmq使用

逃离我推掉我的手 2022-12-09 01:53 146阅读 0赞

# rabbitmq使用 #

## （1）常规 ##

使用一个配置类，在程序启动的时候创建交换机，队列，和绑定队列

@Component
    public class MqConfig { 
    
        /** * 交换机的名字 */
        @Value("${mq.directExchange}")
        private String directExchange;
    
        /** * 队列名 */
        @Value("${mq.queue}")
        private String queue;
    
        /** * 创建一个交换机 * @return */
        @Bean
        public DirectExchange directExchange(){ 
            return new DirectExchange(directExchange);
        }
    
       /** * 创建一个队列 * @return */
        @Bean
        public Queue queue(){ 
            return new Queue(queue,true);
        }
    
     
    
        /** * 绑定队列与交换机 * @return */
        @Bean
        public Binding bindingQueue(){ 
            return BindingBuilder.bind(queue()).to(directExchange()).with(queue);
        }
    }

@Autowired
    private RabbitTemplate rabbitTemplate;

然后就可以在程序中使用了

rabbitTemplate.convertAndSend("交换机名","队列绑定的路由键","消息内容");

## （2）动态 ##

注入操作对象

@Autowired
    private RabbitTemplate rabbitTemplate;

在方法中动态创建交换机，队列，绑定交换机和队列

//创建Rabbit管理器
    RabbitAdmin rabbitAdmin = new RabbitAdmin(rabbitTemplate.getConnectionFactory());
    //创建一个交换机
    DirectExchange directExchange = new DirectExchange("author_subscribe");
    //声明该交换机（只有声明过的交换机才能使用，队列和绑定也是一样）(该交换机用来保存推送的消息）
    rabbitAdmin.declareExchange(directExchange);
    //创建一个队列（用来保存需要推送到某个用户的消息）
    Queue queue = new Queue("author_subscribe_" + userid,true);
    //绑定交换机和队列,设置路由键为作者id（路由键可以绑定多个，发送消息是直接指定交换机和路由键就可以向对应队列发送消息）
    Binding binding = BindingBuilder.bind(queue).to(directExchange).with(articleUserid);

//删除绑定的队列和路由键
    rabbitAdmin.removeBinding(binding);
    //声明队列
    rabbitAdmin.declareQueue(queue);
    //声明绑定
    rabbitAdmin.declareBinding(binding);

## （3）延时队列 ##

@Component
    public class MqConfig { 
    
        /** * 交换机的名字 */
        @Value("${mq.directExchange}")
        private String directExchange;
    
        /** * 接收超时消息的队列名字 */
        @Value("${mq.deadLetterQueue}")
        private String deadLetterQueue;
    
        /** * 延时队列名 */
        @Value("${mq.queue}")
        private String queue;
    
        /** * 创建一个交换机 * @return */
        @Bean
        public DirectExchange directExchange(){ 
            return new DirectExchange(directExchange);
        }
    
        /** * 创建一个延时队列，并设置消息超时后发送的队列 * 消息发送给延时队列，时间到了之后，消息会自动从延时队列发送到死信队列中，这时候我们监听死信队列，就 * 可以知道消息超时了 * @return */
        @Bean
        public Queue queue(){ 
            return QueueBuilder.durable(queue)
                    .withArgument("x-dead-letter-exchange", directExchange)
                    .withArgument("x-dead-letter-routing-key", deadLetterQueue)
                    .build();
        }
    
        /** * 创建一个死信队列，用来接收过期消息 * @return */
        @Bean
        public Queue deadLetterQueue(){ 
            return new Queue(deadLetterQueue,true);
        }
    
    
        /** * 绑定接收过期消息的队列(死信队列）与交换机 * 用户监听死信队列 * @return */
        @Bean
        public Binding bindingQueue(){ 
            return BindingBuilder.bind(deadLetterQueue()).to(directExchange()).with(deadLetterQueue);
        }
    }

## （4）监听（使用注解） ##

@Component
    @RabbitListener(queues = { "监听的队列名"})
    public class messageListener { 
    
        /** * 接收消息 * @param message */
        @RabbitHandler
        public void consumerMessgae(String message){ 
            Map<String,String> resultMap = JSON.parseObject(message, Map.class);
            //message即为mq发送过来的消息
            //下面是对消息的业务处理
            
    
        }
    }

## （5）监听（使用监听容器） ##

某些时候使用注解的方式监听是不符合业务逻辑的，比如监听的队列名是动态的，我们需要需要在接收消息的同时获取到监听的队列名，某些时候队列名中可能包含一些我们需要得到的信息

下面就是自定义的一个监听器

public class SysNoticeListener implements ChannelAwareMessageListener { 
        
        /** * 当该监听器被注册到队列上时，如果有消息到达会触发该方法 * @param message * @param channel * @throws Exception */
        @Override
        public void onMessage(Message message, Channel channel) throws Exception { 
    
            //获取队列名,author_subscribe_1（队列名中包含一些我们需要得到的信息，用户id）
            String consumerQueue = message.getMessageProperties().getConsumerQueue();
             //通过队列名获取用户id
            String userId = consumerQueue.substring(consumerQueue.lastIndexOf("_") + 1);
    
            //根据用户id进行一些业务操作
           
        }
    }

将自定义的监听器注册到mq容器中

@Configuration
    public class RabbitConfig { 
    
        /** * 创建一个容器，设置自定义的监听器 * @param connectionFactory * @return */
        @Bean("sysNoticeContainer")
        public SimpleMessageListenerContainer create(ConnectionFactory connectionFactory){ 
            SimpleMessageListenerContainer simpleMessageListenerContainer =
                    new SimpleMessageListenerContainer(connectionFactory);
            //使用channel
            simpleMessageListenerContainer.setExposeListenerChannel(true);
            //设置自己编写的监听器
            simpleMessageListenerContainer.setMessageListener(new SysNoticeListener());
            //将队列加入到该容器中，是该容器的配置在队列中生效（可以写在外面）
            simpleMessageListenerContainer.addQueueNames("队列名");
            return simpleMessageListenerContainer;
        }
    }

将队列加入到该容器中，是该容器的配置在队列中生效（使用spring的上下文环境，得到该容器）

//使用spring的上下文环境，得到该容器
    SimpleMessageListenerContainer sysNoticeContainer= (SimpleMessageListenerContainer)ApplicationContextProvider.getApplicationContext().getBean("sysNoticeContainer");
    
    
    //在某个方法中动态添加监听
    sysNoticeContainer.addQueueNames("队列名");