springboot：整合rabbitmq之交换机

转x33g5p2x  于2022-06-16 转载在 Spring  
字(13.2k)|赞(0)|评价(0)|浏览(464)

springboot：整合rabbitmq之交换机

环境准备

依赖

  1. <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-web</artifactId>
  4.         </dependency>
  6.         <dependency>
  7.             <groupId>org.springframework.boot</groupId>
  8.             <artifactId>spring-boot-starter-amqp</artifactId>
  9.         </dependency>

配置

  1. @Configuration
  2. public class RabbitMQConfiguration {
  4.     @Bean
  5.     public ConnectionFactory rabbitConnectionFactory() {
  6.         CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
  7.         String rabbitmqHost = "127.0.0.1";
  8.         String rabbitmqPort = "5672";
  9.         String rabbitmqUsername = "guest";
  10.         String rabbitmqPassword = "guest";
  11.         String rabbitmqVirtualHost = "/";
  12.         connectionFactory.setHost(rabbitmqHost);
  13.         connectionFactory.setPort(Integer.parseInt(rabbitmqPort));
  14.         connectionFactory.setUsername(rabbitmqUsername);
  15.         connectionFactory.setPassword(rabbitmqPassword);
  16.         connectionFactory.setVirtualHost(rabbitmqVirtualHost);
  17. //        connectionFactory.setPublisherReturns(true);//开启return模式
  18. //        connectionFactory.setPublisherConfirms(true);//开启confirm模式
  19.         return connectionFactory;
  20.     }
  22.     @Bean(name = "rabbitTemplate")
  23.     //必须是prototype类型
  24.     @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  25.     public RabbitTemplate rabbitTemplate() {
  26.         return new RabbitTemplate(rabbitConnectionFactory());
  27.     }
  29.     @Bean("customContainerFactory")
  30.     public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
  31.         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  32.         //设置线程数
  33.         factory.setConcurrentConsumers(1);
  34.         //最大线程数
  35.         factory.setMaxConcurrentConsumers(1);
  36. //        //设置为手动确认MANUAL(手动),AUTO(自动);
  37. //        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
  38.         // 设置prefetch
  39.         factory.setPrefetchCount(1);
  40.         configurer.configure(factory, connectionFactory);
  41.         return factory;
  42.     }
  43. }

一、交换机

生产者发送消息给交换机，由交换机转发消息给队列，交换机可以转发给所有绑定它的队列，也可以转发给符合路由规则的队列，交换机本身不会存储消息，如果没有绑定任何队列，消息就会丢失

  • 发布订阅模型
    发布订阅使用的交换机是Fanout交换机，也叫广播式交换机
    广播式交换机：fanout交换器中没有路由键的概念，他会把消息发送到所有绑定在此交换器上面的队列
  • 路由模型
    路由式交换机：direct交换器相对来说比较简单，匹配规则为：路由键完全匹配，消息就被投送到相关的队列
  • 主题模型
    主题式交换机：topic交换器采用模糊匹配路由键的原则进行转发消息到队列中
  • 头部订阅
    头部订阅(headers 交换机)：headers没有路由键，是根据消息头部header的键值对进行匹配，可以完全匹配也可以匹配任意一对键值对

二、交换机模式

发布订阅

生产者生产10条消息，发送给FanoutExchange，空字符串""表示不需要指定路由键

消费者A监听队列A，消费者B、B2监听队列B

  1. @RestController
  2. public class FanoutExchangeController {
  4.     @Autowired
  5.     private RabbitTemplate rabbitTemplate;
  7.     @GetMapping("/fanout")
  8.     public void fanout() {
  9.         for (int i = 1; i <= 10; i++) {
  10.             String message = "Hello World ..." + i;
  11.             rabbitTemplate.convertAndSend("fanout_exchange", "", message);
  12.             System.out.println(" [ 生产者 ] Sent ==> '" + message + "'");
  13.         }
  14.     }
  16.     private int count1 = 1;
  17.     private int count2 = 1;
  18.     private int count3 = 1;
  20.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  21.             value = @Queue(value = "fanout_a", declare = "true"),
  22.             exchange = @Exchange(value = "fanout_exchange", type = ExchangeTypes.FANOUT, durable = "true")
  23.     )
  24.     })
  25.     public void receiveA(String message) throws InterruptedException {
  26.         System.out.println(" [ 消费者@A号 ] Received ==> '" + message + "'");
  27.         Thread.sleep(200);
  28.         System.out.println(" [ 消费者@A号 ] Dealt with：" + count1++);
  29.     }
  31.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  32.             value = @Queue(value = "fanout_b", declare = "true"),
  33.             exchange = @Exchange(value = "fanout_exchange", type = ExchangeTypes.FANOUT, durable = "true")
  34.     )
  35.     })
  36.     public void receiveB(String message) throws InterruptedException {
  37.         System.out.println(" [ 消费者@B号 ] Received ==> '" + message + "'");
  38.         Thread.sleep(200);
  39.         System.out.println(" [ 消费者@B号 ] Dealt with：" + count2++);
  40.     }
  42.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  43.             value = @Queue(value = "fanout_b", declare = "true"),
  44.             exchange = @Exchange(value = "fanout_exchange", type = ExchangeTypes.FANOUT, durable = "true")
  45.     )
  46.     })
  47.     public void receiveB2(String message) throws InterruptedException {
  48.         System.out.println(" [ 消费者@B2号 ] Received ==> '" + message + "'");
  49.         Thread.sleep(500);
  50.         System.out.println(" [ 消费者@B2号 ] Dealt with：" + count3++);
  51.     }
  52. }

运行结果

生产者生产了10条消息发送给交换机，fanout交换机将消息分别转发给绑定它的队列A、B；
队列A、B拿到内容一样的10条消息，队列A只有一个消费者A，所以由消费者A处理所有消息
队列B有两个消费者B、B2，分配模式是按能力分配，所以消费者B处理的多

路由模式

  1. @RestController
  2. public class DirectController {
  4.     @Autowired
  5.     private RabbitTemplate rabbitTemplate;
  7.     private int count1 = 1;
  8.     private int count2 = 1;
  9.     private int count3 = 1;
  11.     @GetMapping("/direct")
  12.     public void direct() {
  13.         for (int i = 1; i <= 30; i++) {
  14.             String message = "Hello World ..." + i;
  15.             System.out.println(" [ 生产者 ] Sent ==> '" + message + "'");
  16.             if (i % 3 == 0) {
  17.                 rabbitTemplate.convertAndSend("direct_exchange", "direct_a", message);
  18.             } else if (i % 3 == 1) {
  19.                 rabbitTemplate.convertAndSend("direct_exchange", "direct_b", message);
  20.             } else {
  21.                 rabbitTemplate.convertAndSend("direct_exchange", "direct_b2", message);
  22.             }
  23.         }
  24.     }
  26.     /**
  27.      * name：交换机名称
  28.      * durable：设置是否持久，设置为true则将Exchange存盘，即使服务器重启数据也不会丢失
  29.      * autoDelete：设置是否自动删除，当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange，简单来说也就是如果该Exchange没有和任何队列Queue绑定则删除
  30.      * internal：设置是否是RabbitMQ内部使用，默认false。如果设置为 true ，则表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式。
  31.      * @param message
  32.      * @throws InterruptedException
  33.      */
  34.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  35.             value = @Queue(value = "direct_a", declare = "true"),
  36.             exchange = @Exchange(value = "direct_exchange", type = ExchangeTypes.DIRECT, durable = "true"),
  37.             key = "direct_a"
  38.     )
  39.     })
  40.     public void receiveA(String message) throws InterruptedException {
  41.         System.out.println(" [ 消费者@A号 ] Received ==> '" + message + "'");
  42.         Thread.sleep(200);
  43.         System.out.println(" [ 消费者@A号 ] Dealt with：" + count1++);
  44.     }
  46.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  47.             value = @Queue(value = "direct_b", declare = "true"),
  48.             exchange = @Exchange(value = "direct_exchange", type = ExchangeTypes.DIRECT, durable = "true"),
  49.             key = "direct_b"
  50.     )
  51.     })
  52.     public void receiveB(String message) throws InterruptedException {
  53.         System.out.println(" [ 消费者@B号 ] Received ==> '" + message + "'");
  54.         Thread.sleep(200);
  55.         System.out.println(" [ 消费者@B号 ] Dealt with：" + count2++);
  56.     }
  58.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  59.             value = @Queue(value = "direct_b", declare = "true"),
  60.             exchange = @Exchange(value = "direct_exchange", type = ExchangeTypes.DIRECT, durable = "true"),
  61.             key = "direct_b2"
  62.     )
  63.     })
  64.     public void receiveB2(String message) throws InterruptedException {
  65.         System.out.println(" [ 消费者@B2号 ] Received ==> '" + message + "'");
  66.         Thread.sleep(500);
  67.         System.out.println(" [ 消费者@B2号 ] Dealt with：" + count3++);
  68.     }
  69. }

运行结果

两个队列A、B，A有一个路由键，而B有两个路由键；
生产者生产30条消息发送给交换机，交换机根据路由键转发给队列，队列A分配10条消息，队列B分配20条消息；
同时队列D被两个消费者监听

主题模式(通配者模式)

  1. @RestController
  2. public class TopicController {
  4.     @Autowired
  5.     private RabbitTemplate rabbitTemplate;
  7.     private int count1 = 1;
  8.     private int count2 = 1;
  9.     private int count3 = 1;
  11.     @GetMapping("/topic")
  12.     public void topic() {
  13.         for (int i = 1; i <= 30; i++) {
  14.             String message = "Hello World ..." + i;
  15.             System.out.println(" [ 生产者 ] Sent ==> '" + message + "'");
  16.             if (i % 3 == 0) {
  17.                 // topic.yzm.key，可以匹配 topic.yzm.* 和 topic.#
  18.                 rabbitTemplate.convertAndSend("topic_exchange", "topic.yzm.key", message);
  19.             } else if (i % 3 == 1) {
  20.                 // topic.yzm.yzm，可以匹配 topic.yzm.* 、 topic.# 和 topic.*.yzm
  21.                 rabbitTemplate.convertAndSend("topic_exchange", "topic.yzm.yzm", message);
  22.             } else {
  23.                 // topic.key.yzm，可以匹配 topic.# 和 topic.*.yzm
  24.                 rabbitTemplate.convertAndSend("topic_exchange", "topic.key.yzm", message);
  25.             }
  26.         }
  27.     }
  29.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  30.             value = @Queue(value = "topic_e", declare = "true"),
  31.             exchange = @Exchange(value = "topic_exchange", type = ExchangeTypes.TOPIC, durable = "true"),
  32.             key = "topic.yzm.*"
  33.     )
  34.     })
  35.     public void receiveE(String message) throws InterruptedException {
  36.         System.out.println(" [ 消费者@E号 ] Received ==> '" + message + "'");
  37.         Thread.sleep(200);
  38.         System.out.println(" [ 消费者@E号 ] Dealt with：" + count1++);
  39.     }
  41.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  42.             value = @Queue(value = "topic_f", declare = "true"),
  43.             exchange = @Exchange(value = "topic_exchange", type = ExchangeTypes.TOPIC, durable = "true"),
  44.             key = "topic.#"
  45.     )
  46.     })
  47.     public void receiveF(String message) throws InterruptedException {
  48.         System.out.println(" [ 消费者@F号 ] Received ==> '" + message + "'");
  49.         Thread.sleep(200);
  50.         System.out.println(" [ 消费者@F号 ] Dealt with：" + count2++);
  51.     }
  53.     @RabbitListener(containerFactory = "customContainerFactory", bindings = {@QueueBinding(
  54.             value = @Queue(value = "topic_g", declare = "true"),
  55.             exchange = @Exchange(value = "topic_exchange", type = ExchangeTypes.TOPIC, durable = "true"),
  56.             key = "topic.*.yzm"
  57.     )
  58.     })
  59.     public void receiveG(String message) throws InterruptedException {
  60.         System.out.println(" [ 消费者@G号 ] Received ==> '" + message + "'");
  61.         Thread.sleep(200);
  62.         System.out.println(" [ 消费者@G号 ] Dealt with：" + count3++);
  63.     }
  64. }

分析结果：

E、G二十条，F三十条

头部模式

配置类

  1. @Configuration
  2. public class RabbitMQConfiguration {
  4.     @Bean
  5.     public ConnectionFactory rabbitConnectionFactory() {
  6.         CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
  7.         String rabbitmqHost = "127.0.0.1";
  8.         String rabbitmqPort = "5672";
  9.         String rabbitmqUsername = "guest";
  10.         String rabbitmqPassword = "guest";
  11.         String rabbitmqVirtualHost = "/";
  12.         connectionFactory.setHost(rabbitmqHost);
  13.         connectionFactory.setPort(Integer.parseInt(rabbitmqPort));
  14.         connectionFactory.setUsername(rabbitmqUsername);
  15.         connectionFactory.setPassword(rabbitmqPassword);
  16.         connectionFactory.setVirtualHost(rabbitmqVirtualHost);
  17. //        connectionFactory.setPublisherReturns(true);//开启return模式
  18. //        connectionFactory.setPublisherConfirms(true);//开启confirm模式
  19.         return connectionFactory;
  20.     }
  22.     @Bean(name = "rabbitTemplate")
  23.     //必须是prototype类型
  24.     @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
  25.     public RabbitTemplate rabbitTemplate() {
  26.         return new RabbitTemplate(rabbitConnectionFactory());
  27.     }
  29.     @Bean("customContainerFactory")
  30.     public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
  31.         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  32.         //设置线程数
  33.         factory.setConcurrentConsumers(1);
  34.         //最大线程数
  35.         factory.setMaxConcurrentConsumers(1);
  36. //        //设置为手动确认MANUAL(手动),AUTO(自动);
  37. //        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
  38.         // 设置prefetch
  39.         factory.setPrefetchCount(1);
  40.         configurer.configure(factory, connectionFactory);
  41.         return factory;
  42.     }
  44.     public static final String QUEUE_X = "headers_x";
  45.     public static final String QUEUE_Y = "headers_y";
  46.     public static final String HEADER_EXCHANGE = "headers_exchange";
  48.     @Bean
  49.     public Queue queueX() {
  50.         return new Queue(QUEUE_X);
  51.     }
  53.     @Bean
  54.     public Queue queueY() {
  55.         return new Queue(QUEUE_Y);
  56.     }
  58.     @Bean
  59.     public HeadersExchange headersExchange() {
  60.         return ExchangeBuilder.headersExchange(HEADER_EXCHANGE).build();
  61.     }
  63.     @Bean
  64.     public Binding bindingX() {
  65.         Map<String, Object> map = new HashMap<>();
  66.         map.put("key1", "value1");
  67.         map.put("name", "yzm");
  68.         // whereAll：表示完全匹配
  69.         return BindingBuilder.bind(queueX()).to(headersExchange()).whereAll(map).match();
  70.     }
  72.     @Bean
  73.     public Binding bindingY() {
  74.         Map<String, Object> map = new HashMap<>();
  75.         map.put("key2", "value2");
  76.         map.put("name", "yzm");
  77.         // whereAny：表示只要有一对键值对能匹配就可以
  78.         return BindingBuilder.bind(queueY()).to(headersExchange()).whereAny(map).match();
  79.     }
  80. }

消息发送和接受

  1. @RestController
  2. public class HeadersController {
  4.     @Autowired
  5.     private RabbitTemplate rabbitTemplate;
  7.     @GetMapping("/headers")
  8.     public void headers() {
  9.         String s = "Hello World";
  10.         System.out.println(" [ 生产者 ] Sent ==> '" + s + "'");
  11.         MessageProperties messageProperties = new MessageProperties();
  12.         messageProperties.setHeader("key1", "value1");
  13.         messageProperties.setHeader("name", "yzm");
  14.         Message message = new Message(s.getBytes(StandardCharsets.UTF_8), messageProperties);
  15.         rabbitTemplate.convertAndSend("headers_exchange", "", message);
  16.     }
  18.     @GetMapping("/headers2")
  19.     public void headers2() {
  20.         String s = "Hello World";
  21.         System.out.println(" [ 生产者 ] Sent ==> '" + s + "'");
  22.         MessageProperties messageProperties = new MessageProperties();
  23.         messageProperties.setHeader("key3", "value3");
  24.         messageProperties.setHeader("name", "yzm");
  25.         Message message = new Message(s.getBytes(StandardCharsets.UTF_8), messageProperties);
  26.         rabbitTemplate.convertAndSend("headers_exchange", "", message);
  27.     }
  29.     @RabbitListener(queues = RabbitMQConfiguration.QUEUE_X)
  30.     public void receiveX(String message) {
  31.         System.out.println(" [ 消费者@X号 ] Received ==> '" + message + "'");
  32.     }
  34.     @RabbitListener(queues = RabbitMQConfiguration.QUEUE_Y)
  35.     public void receiveY(String message) {
  36.         System.out.println(" [ 消费者@Y号 ] Received ==> '" + message + "'");
  37.     }
  38. }

测试结果

发送请求：http://localhost:8080/headers

生产消息时，头部键值对有：“key1”=“value1"和"name”=“yzm”，跟X队列能完全匹配上，跟Y队列能匹配上其中一个，所以两个消费者都能消费到消息

发送请求：http://localhost:8080/headers2

只有Y队列能匹配上一个键值对

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_43296313/article/details/125277130

内容来源于网络，如有侵权，请联系作者删除！

Javarabbitmq

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com