RabbitMQ交换机
Posted 嘴强程序员
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ交换机相关的知识,希望对你有一定的参考价值。
1. Exchange作用
在RabbitMQ中,生产者发送消息不会直接将消息投递到队列中,而是先将消息投递到交换机中,在由交换机转发到具体的队列,队列再将消息以推送或者拉取方式给消费者进行消费。
生产者--(创建消息)-->交换机--(路由键)-->队列--(pull/push)-->消费者
2. Exchange的类型
1)直连交换器: Direct Exchange
直连交换机是一种带路由功能的交换机,一个队列会和一个交换机绑定,除此之外再绑定一个routing_key,当消息被发送的时候,需要指定一个binding_key,这个消息被送达交换机的时候,就会被这个交换机送到指定的队列里面去。同样的一个binding_key也是支持应用到多个队列中的
什么是路由键?
每个消息都有一个称为路由键(routing key)的属性,它其实就是一个简单的字符串
直连交换机适用场景?
有优先级的任务,根据任务的优先级把消息发送到对应的队列,这样可以指派更多的资源去处理高优先级的队列。
直连交换机不适合的场景
直连交换机的routing_key方案非常简单,如果我们希望一条消息发送给多个队列,那么直连交换机就不合适了
2)主题交换机: Topic Exchange(发布/订阅)
RabbitMQ提供了一种主题交换机,发送到主题交换机上的消息需要携带指定规则的routing_key,主题交换机会根据这个规则将数据发送到对应的(多个)队列上。
主题交换机的routing_key定义规则:
交换机和队列的binding_key需要采用*.#.*.....的格式,每个部分用.分开,其中:
*表示一个单词
#表示任意数量(零个或多个)单词
示例:
Q1: *.TT.*
Q2: TT.#
如果一条消息携带的路由键为 A.TT.B,那么队列Q1将会收到
如果一条消息携带的路由键为TT.AA.BB,那么队列Q2将会收到
3)广播交换机: Fanout Exchange
用于广播消息,将发送到Exchange中的消息发送到与该交换器关联的所有队列中。
3. 死信队列
死信队列用于存储没匹配队列的消息,超时没有被处理的消息,如果没有配置死信队列这些消息会被丢弃。即当出现没有匹配的队列的消息,或是超时的消息则将消息转入到死信队列里去,等待重新处理或人工干预。
死信队列的应用场景:
- 消息被拒绝(basic.reject或basic.nack)并且requeue=false.
- 消息TTL过期
- 队列达到最大长度(队列满了,无法再添加数据到mq中)
4. 队列参数说明
参数 | 作用 |
---|---|
exchange | 交换机名称 |
type | 交换机类型 |
durable | 是否持久化,RabbitMQ关闭后,没有持久化的Exchange将被清除 |
autoDelete | 是否自动删除,如果没有与之绑定的Queue,直接删除 |
internal | 是否内置的,如果为true,只能通过Exchange到Exchange |
arguments | 结构化参数 |
示例:
Exchange.DeclareOk exchangeDeclare(String exchange,
String type,
boolean durable,
boolean autoDelete,
boolean internal,
Map<String, Object> arguments) throws IOException;
5. 开发示例
准配虚拟机 开启一个Docker 拉取镜像rabbitmq 运行容器
具体步骤:有道云笔记
需要架包
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-test</artifactId>
<scope>test</scope>
</dependency>
配置文件 application 生产与消费者都可用 端口需要改动 还有RabbitMQ服务地址需要改动
server.port=8081
## rabbitmq config
spring.rabbitmq.host=192.168.164.128
spring.rabbitmq.port=5672
spring.rabbitmq.username=xhz
spring.rabbitmq.password=123
spring.rabbitmq.virtual-host=my_vhost
## 消费者数量
spring.rabbitmq.listener.simple.concurrency=10
spring.rabbitmq.listener.simple.max-concurrency=10
#消费者每次从队列中获取的消息数量
spring.rabbitmq.listener.simple.prefetch=1
#消费者自动启动
spring.rabbitmq.listener.simple.auto-startup=true
#消费失败,自动重新入队
spring.rabbitmq.listener.simple.default-requeue-rejected=true
#启用发送重试
spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=1000
spring.rabbitmq.template.retry.max-attempts=3
spring.rabbitmq.template.retry.max-interval=10000
spring.rabbitmq.template.retry.multiplier=1.0
完成演示图
所有消费消息类 模块
5.1 Direct交换机
1)配置直接交换机,队列,并将直接交换机和该队列绑定。(在RabbitMQConfig类中配置,该类使用了@Configuration注解)
package com.rabbitmq.provider.rabbitmqprovider.config;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class DirectConfig
@Bean
public Queue directQueue()
return new Queue("zking-direct-queue");
@Bean
public DirectExchange directExchange()
return new DirectExchange("zking-direct-exchange");
@Bean
public Binding directBinding()
return BindingBuilder.bind(directQueue()).to(directExchange()).with("zking-direc");
2)编写通过直接交换机发送消息的方法
package com.rabbitmq.provider.rabbitmqprovider.web;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
@RestController
public class SenderController
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("/sendDirect")
public String sendDirect(String routing)
Map msg=new HashMap<>();
msg.put("code",200);
msg.put("msg","this message send at time :"+ LocalDateTime.now().format(DateTimeFormatter
.ofPattern("yyyy-MM-dd HH:mm:ss")));
rabbitTemplate.convertAndSend("zking-direct-exchange",routing,msg);
return "direct success";
3.测试交换机发送消息
http://localhost:8081/sendDirect?routing=zking-direc
4.消费消息
创建模块
消费消息 我们运行这个项目
package com.zking.rabbitmq.consumer.rabbitmqconsumer.component;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
@Slf4j
//queues参数指定的是与直接交换机关联的队列名称
@RabbitListener(queues = "zking-direct-queue")
public class DirecReciewer
@RabbitHandler
public void receive(Map msg)
log.info("接收通过直接交换机发送的消息: " + msg);
打印结果
5.2 主题交换机
1) 配置主题交换机,队列,并将主题交换机和该队列绑定。
第一种方式 ---选一种即可
package com.rabbitmq.provider.rabbitmqprovider.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
//@Configuration
public class TopicConfig
/**
* 声明Topic类型的交换机,支持序列化,后面队列进行绑定(topic_queue_q1,topic_queue_q2)
* @return
*/
@Bean(name="topicExchange")
public Exchange topicExchange()
return ExchangeBuilder
.topicExchange("topic_exchange")
.durable(true)
.build();
/**
* 声明队列,该队列与topic交换机绑定
* @return
*/
@Bean(name="topicQueue1")
public Queue topicQueue1()
return QueueBuilder.durable("topic_queue_q1").build();
/**
* 声明队列,该队列与topic交换机绑定
* @return
*/
@Bean(name="topicQueue2")
public Queue topicQueue2()
return QueueBuilder.durable("topic_queue_q2").build();
/**
* 将队列(topic_queue_q1)与topic型交换机进行绑定
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding topicBindingQ1(
@Qualifier("topicQueue1") Queue queue,
@Qualifier("topicExchange") Exchange exchange)
return BindingBuilder
.bind(queue)
.to(exchange)
.with("topic.queue.#")
.noargs();
/**
* 将队列(topic_queue_q2)与topic型交换机进行绑定
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding topicBindingQ2(
@Qualifier("topicQueue2") Queue queue,
@Qualifier("topicExchange") Exchange exchange)
return BindingBuilder
.bind(queue)
.to(exchange)
.with("topic.queue.#")
.noargs();
测试 发送消息
package com.rabbitmq.provider.rabbitmqprovider;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
@SpringBootTest
class RabbitmqProviderApplicationTests
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
void contextLoads()
Map msg=new HashMap<>();
msg.put("code",200);
msg.put("msg","this message send at time :"+ LocalDateTime.now().format(DateTimeFormatter
.ofPattern("yyyy-MM-dd HH:mm:ss")));
rabbitTemplate.convertAndSend("topic_exchange","topic.queue.ab",msg);
第二种发送消息
package com.rabbitmq.provider.rabbitmqprovider.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class TopicConfig1
@Bean(name="topicQueue1")
public Queue topicQueue1()
return QueueBuilder.durable("topic_queue_q1").build();
@Bean(name="topicQueue2")
public Queue topicQueue2()
return QueueBuilder.durable("topic_queue_q2").build();
@Bean
public TopicExchange topicExchange()
return new TopicExchange("topic-exchange");
@Bean
public Binding topicBinding1( @Qualifier("topicQueue1") Queue queue,
@Qualifier("topicExchange") TopicExchange exchange)
return BindingBuilder.bind(queue).to(exchange).with("person.yy");
@Bean
public Binding topicBinding2( @Qualifier("topicQueue2") Queue queue,
@Qualifier("topicExchange") TopicExchange exchange)
return BindingBuilder.bind(queue).to(exchange).with("person.*");
测试发送消息
package com.rabbitmq.provider.rabbitmqprovider.web;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
@RestController
public class SenderController
@RequestMapping("/sendTopic")
public String sendTopic(String routing)
Map msg=new HashMap<>();
msg.put("code",200);
msg.put("msg","this message send at time :"+ LocalDateTime.now().format(DateTimeFormatter
.ofPattern("yyyy-MM-dd HH:mm:ss")));
rabbitTemplate.convertAndSend("topic-exchange",routing,msg);
return "direct success";
http://localhost:8081/sendTopic?routing=person.y 只有条件为 y 或者 *
http://localhost:8081/sendTopic?routing=person.yy
消费消息
package com.zking.rabbitmq.consumer.rabbitmqconsumer.component;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
public class TopicReciewer
@RabbitListener(queues="topic_queue_q1")
@RabbitHandler
public void handler(Map map)
System.out.println(map);
@RabbitListener(queues="topic_queue_q2")
@RabbitHandler
public void handler1(Map map)
System.out.println(map);
5.3 广播交换机 (扇形)
1)配置广播交换机,队列,并将主题交换机和该队列绑定。
package com.rabbitmq.provider.rabbitmqprovider.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class FanoutConfig
@Bean
public Queue fanoutQueue1()
return new Queue("fanout-queue1");
@Bean
public Queue fanoutQueue2()
return new Queue("fanout-queue2");
@Bean
public Queue fanoutQueue3()
return new Queue("fanout-queue3");
@Bean
public FanoutExchange fanoutExchange()
return new FanoutExchange("fanout-exchange");
@Bean
public Binding fanoutBInding1(@Qualifier("fanoutQueue1") Queue queue,
@Qualifier("fanoutExchange") FanoutExchange exchange)
return BindingBuilder.bind(queue).to(exchange);
@Bean
public Binding fanoutBInding2(@Qualifier("fanoutQueue2") Queue queue,
@Qualifier("fanoutExchange") FanoutExchange exchange)
return BindingBuilder.bind(queue).to(exchange);
@Bean
public Binding fanoutBInding3(@Qualifier("fanoutQueue3") Queue queue,
@Qualifier("fanoutExchange") FanoutExchange exchange)
return BindingBuilder.bind(queue).to(exchange);
生产的队列
向服务器发送消息
package com.rabbitmq.provider.rabbitmqprovider.web;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
@RestController
public class SenderController
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("/sendFanout")
public String sendFanout()
Map msg=new HashMap<>();
msg.put("code",200);
msg.put("msg","this message send at time :"+ LocalDateTime.now().format(DateTimeFormatter
.ofPattern("yyyy-MM-dd HH:mm:ss")));
rabbitTemplate.convertAndSend("fanout-exchange",null,msg);
return "direct success";
发送与消费
消费消息
package com.zking.rabbitmq.consumer.rabbitmqconsumer.component;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
public class FanoutRecevier
@RabbitListener(queues="fanout-queue1")
@RabbitHandler
public void fanout(Map map)
System.out.println(map);
@RabbitListener(queues="fanout-queue2")
@RabbitHandler
public void fanout1(Map map)
System.out.println(map);
@RabbitListener(queues="fanout-queue3")
@RabbitHandler
public void fanout2(Map map)
System.out.println(map);
以上是关于RabbitMQ交换机的主要内容,如果未能解决你的问题,请参考以下文章