devbox
小蓝xlanll
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

消息队列中间件，RabbitMQ的使用，死信队列，延迟队列，利用枚举实现队列，交换机，RountKey的声明

作者：小蓝xlanll | 2024-04-16 22:26:20

赞

踩

消息队列中间件，RabbitMQ的使用，死信队列，延迟队列，利用枚举实现队列，交换机，RountKey的声明

目录

0.交换机种类和区别

1.声明队列和交换机以及RountKey

2.初始化循环绑定

3.声明交换机

4.监听队列

4.1 监听普通队列

4.2监听死信队列

 5.削峰填谷的实现

0.交换机种类和区别

  1. Direct Exchange（直连交换机）：

    • 直连交换机将消息发送到与消息的路由键完全匹配的队列。它是最简单的交换机类型之一。
    • 当一个队列使用某个直连交换机绑定时，它需要指定一个绑定键（binding key），当消息的路由键与该绑定键完全匹配时，消息会被发送到该队列。

  2. Fanout Exchange（扇出交换机）：

    • 扇出交换机会将消息发送到与其绑定的所有队列，忽略消息的路由键。
    • 当一个队列使用扇出交换机绑定时，它会接收到交换机发送的所有消息，无论消息的路由键是什么。

  3. Topic Exchange（主题交换机）：

    • 主题交换机根据消息的路由键和绑定键之间的模式匹配来路由消息。
    • 绑定键可以使用通配符进行匹配，支持 '*' 匹配一个单词，'#' 匹配零个或多个单词，从而允许更灵活的路由规则。

  4. Headers Exchange（标头交换机）：

    • 标头交换机根据消息的标头（headers）中的键值对来路由消息，而不是使用路由键。
    • 在将队列绑定到标头交换机时，可以指定一组标头键值对，只有当消息的标头中包含与绑定相匹配的所有键值对时，消息才会被路由到该队列。

如果满足key的前提下，绑定同一个交换机的队列都会分配到相同数量的信息

比如此时交换机有20条信息，a，b队列都会分配到20条信息

默认情况下，会轮询分配给消费者，也可以设置最多获取多少条未被消费的信息，根据消费者的消费能力来设置

1.声明队列和交换机以及RountKey

  1. package com.example.config;
  2.  
  3.  
  4. import lombok.Getter;
  5.  
  6. @Getter
  7. public enum RabbitmqBind {
  8.  
  9.  
  10.     DATA_CLEAN_PROCESS_DEAD(
  11.             RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP,
  12.             RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS_DEAD,
  13.             RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS_DEAD,
  14.             false,
  15.             false,
  16.             null,
  17.             null
  18.     ),
  19.  
  20.     DATA_CLEAN_PROCESS(
  21.             RabbitMqExchangeEnum.E_DIRECT_RCP,
  22.             RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS,
  23.             RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS,
  24.             true,
  25.             true,
  26.             RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP,
  27.             RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS_DEAD),
  28.  
  29.     SMS_CLEAN_DEAD(
  30.             RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP,
  31.             RabbitMqQueueConstants.Q_API_TO_DCN_SMS_DEAD,
  32.             RabbitmqRoutingKey.K_DATA_CLEAN_FINISH_DEAD,
  33.             true,
  34.             false,
  35.             null,
  36.             null
  37.     ),
  38.  
  39.     SMS_CLEAN(
  40.             RabbitMqExchangeEnum.E_TOPIC_RCP,
  41.             RabbitMqQueueConstants.Q_API_TO_DCN_SMS,
  42.             RabbitmqRoutingKey.K_API_TO_DCN_SMS,
  43.             true,
  44.             true,
  45.             RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP,
  46.             RabbitmqRoutingKey.K_DATA_CLEAN_FINISH_DEAD
  47.     ),
  48.  
  49.  
  50.     ;
  51.  
  52.     /**
  53.      * 交换机
  54.      */
  55.     private RabbitMqExchangeEnum exchange;
  56.  
  57.     /**
  58.      * 队列名称
  59.      */
  60.     private String queueName;
  61.  
  62.     /**
  63.      * 路由Key
  64.      */
  65.     private RabbitmqRoutingKey routingKey;
  66.  
  67.     /**
  68.      * 绑定标识
  69.      * 是否启用
  70.      */
  71.     private Boolean isBind;
  72.  
  73.     /**
  74.      * 是否绑定死信
  75.      */
  76.     private Boolean isDeathBelief;
  77.  
  78.     /**
  79.      * 绑定的死信交换机
  80.      */
  81.     private RabbitMqExchangeEnum boundDeadExchange;
  82.  
  83.     /**
  84.      * 死信key
  85.      */
  86.     private RabbitmqRoutingKey deadRoutingKey;
  87.  
  88.  
  89.     RabbitmqBind(RabbitMqExchangeEnum exchange, String queueName, RabbitmqRoutingKey routingKey, Boolean isBind,
  90.                  Boolean isDeathBelief, RabbitMqExchangeEnum boundDeadExchange, RabbitmqRoutingKey deadRoutingKey
  91.     ) {
  92.         this.exchange = exchange;
  93.         this.queueName = queueName;
  94.         this.routingKey = routingKey;
  95.         this.isBind = isBind;
  96.         this.isDeathBelief = isDeathBelief;
  97.         this.boundDeadExchange = boundDeadExchange;
  98.         this.deadRoutingKey = deadRoutingKey;
  99.     }
  100.  
  101.     /**
  102.      * 交换机
  103.      */
  104.     @Getter
  105.     public enum RabbitMqExchangeEnum {
  106.  
  107.         /**
  108.          * 交换机定义，类型 - 名称
  109.          */
  110.         E_DIRECT_RCP("direct", "E_DIRECT_RCP"),
  111.         DEAD_E_DIRECT_RCP("direct", "DEAD_E_DIRECT_RCP"),
  112.  
  113.         E_TOPIC_RCP("topic", "E_TOPIC_RCP"),
  114.  
  115.         E_TOPIC_PAY("topic", "E_TOPIC_PAY");
  116.  
  117.         private String exchangeType;
  118.  
  119.         private String exchangeName;
  120.  
  121.         RabbitMqExchangeEnum(String exchangeType, String exchangeName) {
  122.             this.exchangeType = exchangeType;
  123.             this.exchangeName = exchangeName;
  124.         }
  125.     }
  126.  
  127.     /**
  128.      * 队列名定义
  129.      */
  130.     public interface RabbitMqQueueConstants {
  131.  
  132.         /**
  133.          * 接收清洗数据
  134.          */
  135.         String Q_DATA_CLEAN_PROCESS = "RMPS_TO_RCP_DATA_CLEAN_PROCESS";
  136.  
  137.         /**
  138.          * 清洗结束通知
  139.          */
  140.         String Q_API_TO_DCN_SMS = "Q_API_TO_DCN_SMS";
  141.  
  142.         /**
  143.          * 死信队列
  144.          */
  145.         String Q_DATA_CLEAN_PROCESS_DEAD = "Q_DATA_CLEAN_PROCESS_DEAD";
  146.  
  147.         /**
  148.          * 清洗结束通知死信队列
  149.          */
  150.         String Q_API_TO_DCN_SMS_DEAD = "Q_API_TO_DCN_SMS_DEAD";
  151.     }
  152.  
  153.     /**
  154.      * routingKey
  155.      */
  156.     @Getter
  157.     public enum RabbitmqRoutingKey {
  158.  
  159.         /**
  160.          * 路由
  161.          */
  162.         K_DATA_CLEAN_PROCESS("K_DATA_CLEAN_PROCESS"),
  163.         K_API_TO_DCN_SMS("K_API_TO_DCN_SMS"),
  164.  
  165.         // 路由绑定死信路由
  166.         DEAD("DEAD"),
  167.  
  168.         //死信路由
  169.         K_DATA_CLEAN_PROCESS_DEAD("K_DATA_CLEAN_PROCESS_DEAD"),
  170.         K_DATA_CLEAN_FINISH_DEAD("K_DATA_CLEAN_FINISH_DEAD"),
  171.         ;
  172.  
  173.         private String keyName;
  174.  
  175.         RabbitmqRoutingKey(String keyName) {
  176.             this.keyName = keyName;
  177.         }
  178.     }
  179.  
  180. }

2.初始化循环绑定

  1. package com.example.config;
  2.  
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.Exchange;
  5. import org.springframework.amqp.core.Queue;
  6. import org.springframework.amqp.core.QueueBuilder;
  7. import org.springframework.amqp.rabbit.annotation.EnableRabbit;
  8. import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
  9. import org.springframework.amqp.rabbit.connection.ConnectionFactory;
  10. import org.springframework.amqp.rabbit.core.RabbitAdmin;
  11. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  12. import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
  13. import org.springframework.beans.factory.annotation.Autowired;
  14. import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
  15. import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
  16. import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
  17. import org.springframework.context.annotation.Bean;
  18. import org.springframework.context.annotation.Configuration;
  19. import org.springframework.context.annotation.Lazy;
  20.  
  21. import javax.annotation.PostConstruct;
  22. import javax.annotation.Resource;
  23. import java.util.Arrays;
  24.  
  25. @Configuration
  26. @ConditionalOnClass(EnableRabbit.class)
  27. public class MqConfig {
  28.     @Resource
  29.     protected RabbitTemplate rabbitTemplate;
  30.     @Resource
  31.     ConnectionFactory connectionFactory;
  32. //
  33. //    @Lazy
  34. //    @Autowired
  35. //    protected RabbitAdmin rabbitAdmin;
  36. //
  37. //
  38. //    public static final int DEFAULT_CONCURRENT = 10;
  39. //
  40. //    @Bean("customContainerFactory")
  41. //    public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer,
  42. //                                                                 ConnectionFactory connectionFactory) {
  43. //        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  44. //        factory.setConcurrentConsumers(DEFAULT_CONCURRENT);
  45. //        factory.setMaxConcurrentConsumers(DEFAULT_CONCURRENT);
  46. //        configurer.configure(factory, connectionFactory);
  47. //        return factory;
  48. //    }
  49. //
  50. //    @Bean
  51. //    @ConditionalOnMissingBean
  52. //    public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
  53. //        return new RabbitTransactionManager(connectionFactory);
  54. //    }
  55. //
  56. //    @Bean
  57. //    @ConditionalOnMissingBean
  58. //    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
  59. //        return new RabbitAdmin(connectionFactory);
  60. //    }
  61.  
  62.     @PostConstruct
  63.     protected void init() {
  64.         RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
  65.  
  66.  
  67.         rabbitTemplate.setChannelTransacted(true);
  68.         //创建exchange
  69.         Arrays.stream(RabbitmqBind.RabbitMqExchangeEnum.values())
  70.                 .forEach(rabbitMqExchangeEnum -> {
  71.                             Exchange exchange = RabbitmqExchange
  72.                                     .getInstanceByType(rabbitMqExchangeEnum.getExchangeType())
  73.                                     .createExchange(rabbitMqExchangeEnum.getExchangeName());
  74.                             rabbitAdmin.declareExchange(exchange);
  75.                         }
  76.                 );
  77.  
  78.         //创建队列并绑定exchange
  79.         Arrays.stream(RabbitmqBind.values()).forEach(RabbitmqBind -> {
  80.             if (RabbitmqBind.getIsBind()) {
  81.                 if (RabbitmqBind.getIsDeathBelief()) {
  82.                     //需要绑定死信交换机的队列
  83.                     rabbitAdmin.declareQueue(QueueBuilder.durable(RabbitmqBind.getQueueName())
  84.                             .ttl(60000).deadLetterExchange(RabbitmqBind.getBoundDeadExchange().getExchangeName())
  85.                             .deadLetterRoutingKey(RabbitmqBind.getDeadRoutingKey().getKeyName()).build());
  86.                     rabbitAdmin.declareBinding(new Binding(RabbitmqBind.getQueueName(),
  87.                             Binding.DestinationType.QUEUE,
  88.                             RabbitmqBind.getExchange().getExchangeName(),
  89.                             RabbitmqBind.getRoutingKey().getKeyName(), null));
  90.                 } else {
  91.                     //不需要绑定死信交换机的队列
  92.                     rabbitAdmin.declareQueue(new Queue(RabbitmqBind.getQueueName(),
  93.                             true, false, false, null));
  94.                     rabbitAdmin.declareBinding(new Binding(RabbitmqBind.getQueueName(),
  95.                             Binding.DestinationType.QUEUE,
  96.                             RabbitmqBind.getExchange().getExchangeName(),
  97.                             RabbitmqBind.getRoutingKey().getKeyName(), null));
  98.                 }
  99.             }
  100.         });
  101.     }
  102.  
  103. }

 绑定的形式由枚举类中定义

3.声明交换机

  1. package com.example.config;
  2.  
  3. import lombok.Getter;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.springframework.amqp.core.DirectExchange;
  6. import org.springframework.amqp.core.Exchange;
  7. import org.springframework.amqp.core.TopicExchange;
  8.  
  9. import java.util.Arrays;
  10.  
  11.  
  12. @Getter
  13. @Slf4j
  14. public enum RabbitmqExchange {
  15.  
  16.     DIRECT("direct"){
  17.         @Override
  18.         public Exchange createExchange(String exchangeName) {
  19.             return new DirectExchange(exchangeName, true, false);
  20.         }
  21.     },
  22.  
  23.     TOPIC("topic"){
  24.         @Override
  25.         public Exchange createExchange(String exchangeName) {
  26.             return new TopicExchange(exchangeName, true, false);
  27.         }
  28.     };
  29.  
  30.     public static RabbitmqExchange getInstanceByType(String type){
  31.  
  32.         return Arrays.stream(RabbitmqExchange.values()).filter(e -> e.getType().equals(type))
  33.                 .findAny()
  34.                 .orElseThrow(() ->
  35. //                        new ProcessException("无效的exchange type")
  36.  
  37.                         new RuntimeException("无效的exchange type")
  38.                 );
  39.     }
  40.  
  41.     private String type;
  42.  
  43.  
  44.     RabbitmqExchange(String type) {
  45.         this.type = type;
  46.     }
  47.  
  48.     public abstract Exchange createExchange(String exchangeName);
  49.  
  50. }

4.监听队列

4.1 监听普通队列

  1. package com.example.listener;
  2.  
  3. import com.example.config.RabbitmqBind;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.commons.lang.StringUtils;
  6. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  7. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  8. import org.springframework.stereotype.Component;
  9.  
  10.  
  11. @Slf4j
  12. @Component
  13. @RabbitListener(queues = {
  14.         RabbitmqBind.RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS }, concurrency = "1-5")
  15. //, containerFactory = "customContainerFactory"
  16. public class MqListener {
  17.  
  18.  
  19.  
  20.     @RabbitHandler
  21.     public void processMessage(String message) {
  22.         log.info("DataClean recive message :{} ", message);
  23.         process(message);
  24.     }
  25.  
  26.     @RabbitHandler
  27.     public void processMessage(byte[] message) {
  28.         String msg = new String(message);
  29.         log.info("DataClean recive message :{} ", msg);
  30.         process(msg);
  31.     }
  32.  
  33.     /**
  34.      * 处理推送消息
  35.      * @param message
  36.      */
  37.     private void process(String message) {
  38.         log.info("process message :{}" , message);
  39.         if(StringUtils.isBlank(message)) {
  40.             log.error("process message is blank , message:{}" , message);
  41.             return;
  42.         }
  43.     }
  44.  
  45. }

 监听并处理任务

4.2监听死信队列

  1. package com.example.listener;
  2.  
  3. import com.example.config.RabbitmqBind;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.apache.commons.lang.StringUtils;
  6. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
  7. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  8. import org.springframework.stereotype.Component;
  9.  
  10. @Slf4j
  11. @Component
  12. @RabbitListener(queues = {
  13.         RabbitmqBind.RabbitMqQueueConstants.Q_API_TO_DCN_SMS_DEAD }, concurrency = "1-5")
  14. public class DeadListener {
  15.  
  16.     @RabbitHandler
  17.     public void processMessage(String message) {
  18.         log.info("DataClean recive message :{} ", message);
  19.         process(message);
  20.     }
  21.  
  22.     @RabbitHandler
  23.     public void processMessage(byte[] message) {
  24.         String msg = new String(message);
  25.         log.info("DataClean recive message :{} ", msg);
  26.         process(msg);
  27.     }
  28.  
  29.  
  30.     /**
  31.      * 处理推送消息
  32.      * @param message
  33.      */
  34.     private void process(String message) {
  35.         log.info("Dead process message :{}" , message);
  36.         if(StringUtils.isBlank(message)) {
  37.             log.error("Dead process message is blank , message:{}" , message);
  38.             return;
  39.         }
  40.     }
  41.  
  42. }

 5.削峰填谷的实现

把高峰期的消息填进低峰期

可以用拉取的方式来实现

或者用消费者的最大数量和最小数量来实现

channel.basicQos();//设置最大获取未确认消息的数量，实现权重

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/436766
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号