zoukankan      html  css  js  c++  java
  • RabbitMQ延迟消息学习

    准备做一个禁言自动解除的功能,立马想到了订单的超时自动解除,刚好最近在看RabbitMQ的实现,于是想用它实现,查询了相关文档发现确实可以实现,动手编写了这篇短文。

    准备工作

    1、Erlang安装请参考windows下安装Erlang
    2、mq安装晴参考RabbitMQ安装
    3、延迟消息插件安装rabbitmq_delayed_message_exchange

        #插件下载地址(选择与mq版本匹配的插件版本)
        http://www.rabbitmq.com/community-plugins.html
        #安装命令如下(在安装目录sbin下执行如下命令)
        rabbitmq-plugins enable rabbitmq_delayed_message_exchange
    

    创建项目

    我选择的是在springboot中集成RabbitMQ,配置相对简单很多。

    项目创建好后,在application.properties中加入RabbitMQ参数:

    #RabbitMQ config
    spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    #Custom config
    rabbitmq.exchange=test_exchange
    rabbitmq.queue=test_queue_1
    

    定义ConnectionFactory和RabbitTemplate

        package com.xsh.mq.config;
    
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    @ConfigurationProperties(prefix = "spring.rabbitmq")
    public class RabbitMqConfig {
        private String host;
        private int port;
        private String userName;
        private String password;
    
        @Bean
        public ConnectionFactory connectionFactory() {
            CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(host,port);
            cachingConnectionFactory.setUsername(userName);
            cachingConnectionFactory.setPassword(password);
            cachingConnectionFactory.setVirtualHost("/");
            cachingConnectionFactory.setPublisherConfirms(true);
            return cachingConnectionFactory;
        }
    
        @Bean
        public RabbitTemplate rabbitTemplate() {
            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
            return rabbitTemplate;
        }
    
        public String getHost() {
            return host;
        }
    
        public void setHost(String host) {
            this.host = host;
        }
    
        public int getPort() {
            return port;
        }
    
        public void setPort(int port) {
            this.port = port;
        }
    
        public String getUserName() {
            return userName;
        }
    
        public void setUserName(String userName) {
            this.userName = userName;
        }
    
        public String getPassword() {
            return password;
        }
    
        public void setPassword(String password) {
            this.password = password;
        }
    }
    

    Exchange和Queue配置

        package com.xsh.mq.config;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.CustomExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 配置队列
     */
    @Configuration
    public class QueueConfig {
    
        @Value("${rabbitmq.exchange}")
        private String exchangeName;
    
        @Value("${rabbitmq.queue}")
        private String queueName;
    
        @Bean
        public CustomExchange delayExchange() {
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct");
           //使用的是CustomExchange,不是DirectExchange,另外CustomExchange的类型必须是x-delayed-message
            return new CustomExchange(exchangeName, "x-delayed-message",true, false,args);
        }
    
        @Bean
        public Queue queue() {
            Queue queue = new Queue(queueName, true);
            return queue;
        }
    
        @Bean
        public Binding binding() {
            return BindingBuilder.bind(queue()).to(delayExchange()).with(queueName).noargs();
        }
    }
    

    消息发送

        package com.xsh.mq.service;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    
    @Service
    public class MessageServiceImpl {
    
        /**
         * 日志
         */
        private static final Logger logger = LoggerFactory.getLogger(MessageServiceImpl.class);
        /**
         * rabbitMQ模板
         */
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Value("${rabbitmq.exchange}")
        private String exchangeName;
    
        /**
         * 发送消息
         * @param queueName 队列名称
         * @param msg 消息内容
         * @param delay 延迟时长 默认3秒
         */
        public void sendMsg(String queueName,String msg,Integer delay) {
            if(null == delay){
                delay = 3000;
            }
            logger.info("》》》》发送消息");
            Integer finalDelay = delay;
            rabbitTemplate.convertAndSend(exchangeName, queueName, msg, message -> {
                //必须添加header x-delay
                message.getMessageProperties().setHeader("x-delay", finalDelay);
                return message;
            });
        }
    }
    

    这里发送消息我定义了一个延迟参数,传入的延迟是多少,消息就延迟多少,方便消息延迟不一样

    消费消息

        package com.xsh.mq.service;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    @Component
    public class MessageReceiver {
        /**
         * 日志
         */
        private static final Logger logger = LoggerFactory.getLogger(MessageReceiver.class);
    
        @RabbitListener(queues = "${rabbitmq.queue}")
        public void receive(String msg) {
            logger.info("收到消息:{}", msg);
        }
    }
    

    测试发送接收

    先运行springboot项目,然后编写单元测试用例

          package com.xsh.mq;
    
      import com.xsh.mq.service.MessageServiceImpl;
      import org.junit.Test;
      import org.junit.runner.RunWith;
      import org.springframework.beans.factory.annotation.Autowired;
      import org.springframework.beans.factory.annotation.Value;
      import org.springframework.boot.test.context.SpringBootTest;
      import org.springframework.test.context.junit4.SpringRunner;
    
      @RunWith(SpringRunner.class)
      @SpringBootTest
      public class MqApplicationTests {
    
          @Test
          public void contextLoads() {
          }
    
          @Autowired
          private MessageServiceImpl messageService;
    
          @Value("${rabbitmq.queue}")
          private String queueName;
    
          @Test
          public void send() {
              messageService.sendMsg(queueName, "delayMsg2", 1000 * 60 * 2);
              messageService.sendMsg(queueName, "delayMsg1", 1000 * 60);
              messageService.sendMsg(queueName, "delayMsg3", 1000 * 60*3);
          }
    
      }
    

    这里我发送了三条延迟消息,控制台结果如图:

    消费者接收到的消息为:

    从执行结果来看,demo基本实现,RabbitMQ其他细节还有待继续看。
    参考文章:Scheduling Messages with RabbitMQ

  • 相关阅读:
    449. 序列化和反序列化二叉搜索树
    极客mysql08
    极客mysql06
    MySQL死锁问题(转)
    mysql一些好的问题
    如何在Ubuntu 20.04 LTS Focal Fossa上安装Dokuwiki
    在Ubuntu 20.04 LTS Focal Fossa上安装Icinga
    如何在Ubuntu 20.04 LTS Focal Fossa服务器上安装PowerShell
    如何在CentOS 8服务器上安装Discord
    在CentOS 8服务器上安装Apache Solr搜索平台
  • 原文地址:https://www.cnblogs.com/xieshuang/p/10121891.html
Copyright © 2011-2022 走看看