柚子快报激活码778899分享：RabbitMQ 学习笔记

柚子快报激活码778899分享：RabbitMQ 学习笔记

http://yzkb.51969.com/

为什么使用消息队列？

以用户下单购买商品的行为举例，在使用微服务架构时，我们需要调用多个服务，传统的调用方式是同步调用，这会存在一定的性能问题

使用消息队列可以实现异步的通信方式，相比于同步的通信方式，异步的方式可以让上游快速成功，极大提高系统的吞吐量

消息队列的使用场景有如下：

异步处理：以上述用户下单购买商品为例，将多个不关联的任务放进消息队列，提高系统性能应用解耦：以上述用户下单购买商品为例，订单系统通知库存系统减库存，传统的做法是订单系统调用库存系统的接口，订单系统和库存系统高耦合，当库存系统出现故障时，订单就会失败。使用消息队列，用户下单后，订单系统完成持久化，将消息写入消息队列，返回用户下单成功。库存系统订阅下单消息，获取下单消息，进行减库存操作。就算库存系统出现故障，消息队列也能保证消息的可靠投递，不会导致系统异常流量削峰：举行秒杀活动时，为防止流量过大导致应用挂掉，服务器收到用户请求后，先写入消息队列，如果超过了消息队列长度的最大值，则直接抛弃或跳转到错误页面。秒杀业务根据消息队列中的请求信息，再做后续处理，缓解短时间的流量高峰

RabbitMQ 安装

以 ubuntu 22.04.3 为例，参考 RabbitMQ 官网提供的安装脚本

#!/bin/sh

## 要想安装最新版本的 rabbitmq，可以选择 Cloudsmith 存储库下载，为此我们必须安装 apt-transport https 包

sudo apt-get install curl gnupg apt-transport-https -y

## 获取 Cloudsmith 存储库提供的签名密钥并添加到系统中，这样这样才能使用 Cloudsmith 仓库下载包

curl -1sLf "https://keys.openpgp.org/vks/v1/by-fingerprint/0A9AF2115F4687BD29803A206B73A36E6026DFCA" | sudo gpg --dearmor | sudo tee /usr/share/keyrings/com.rabbitmq.team.gpg > /dev/null

curl -1sLf https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.E495BB49CC4BBE5B.key | sudo gpg --dearmor | sudo tee /usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg > /dev/null

curl -1sLf https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9F4587F226208342.key | sudo gpg --dearmor | sudo tee /usr/share/keyrings/rabbitmq.9F4587F226208342.gpg > /dev/null

## 将描述 RabbitMQ 和 Erlang 包存储库的文件放在 /etc/apt/sources.list.d/ 目录下

sudo tee /etc/apt/sources.list.d/rabbitmq.list <

## Provides modern Erlang/OTP releases

##

deb [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa1.novemberain.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main

deb-src [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa1.novemberain.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main

# another mirror for redundancy

deb [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa2.novemberain.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main

deb-src [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa2.novemberain.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main

## Provides RabbitMQ

##

deb [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa1.novemberain.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main

deb-src [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa1.novemberain.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main

# another mirror for redundancy

deb [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa2.novemberain.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main

deb-src [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa2.novemberain.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main

EOF

## 更新存储库索引

sudo apt-get update -y

## 安装 erlang

sudo apt-get install -y erlang-base \

erlang-asn1 erlang-crypto erlang-eldap erlang-ftp erlang-inets \

erlang-mnesia erlang-os-mon erlang-parsetools erlang-public-key \

erlang-runtime-tools erlang-snmp erlang-ssl \

erlang-syntax-tools erlang-tftp erlang-tools erlang-xmerl

## 安装 rabbitmq

sudo apt-get install rabbitmq-server -y --fix-missing

使用以下命令启动、关闭和查看 rabbitmq 状态

sudo systemctl stop rabbitmq-server

sudo systemctl start rabbitmq-server

sudo systemctl status rabbitmq-server

要想访问 rabbitmq 的 web 管理界面，需要执行以下命令，启动 rabbitmq 的插件管理

rabbitmq-plugins enable rabbitmq_management

访问：http://127.0.0.1:15672/ 可查看 rabbitmq 的 web 管理界面，但首先要创建用户，这里创建一个管理员用户，使用该用户登录

# rabbitmqctl add_user {username} {password}

# 设置账号和密码

rabbitmqctl add_user admin 123

# rabbitmqctl set_user_tags {username} {role}

# 设置角色，administrator 是管理员角色

rabbitmqctl set_user_tags admin

# rabbitmqctl set_permissions [-p vhost] {user} {conf} {write} {read}

# 设置权限：

# {vhost} 表示待授权用户访问的 vhost 名称，默认为 "/"

# {user} 表示待授权访问特定 vhost 的用户名称

# {conf} 表示待授权用户的配置权限，是一个匹配资源名称的正则表达式

# {write} 表示待授权用户的写权限，是一个匹配资源名称的正则表达式

# {read} 表示待授权用户的读权限，是一个资源名称的正则表达式

rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

更多 rabbitmqctl 命令可参考官网：https://www.rabbitmq.com/rabbitmqctl.8.html

RabbitMQ 组成与消息模型

RabbitMQ 是使用 Erlang 语言开发的开源的消息队列，基于 AMQP（高级消息队列协议）实现

RabbitMQ 的组成部分如下：

Message：消息，又可分为消息头和消息体，消息头由一系列可选属性组成Producer：消息生产者，是向交换器发布消息的客户端应用程序Consumer：消息消费者，从消息队列获取消息的客户端应用程序Exchange：交换器，接收生产者发送的消息并路由到相应的队列，常用的交换器类型有 direct、fanout、topicBinding：绑定，用于关联消息队列和交换器Queue：消息队列，保存消息直到放送给消费者Rounting-key：路由键，决定信息该投递到哪个队列的规则Connection：链接，指应用与 rabbit 服务器建立的 TCP 链接Channel：信道，TCP 里面的虚拟链接，一条 TCP 链接上可以创建多条信道，可以避免频繁创建和销毁 TCP 连接所带来的开销Virtual Host：虚拟主机，表示一批交换器、消息队列和相关对象，是共享相同身份认证和加密环境的独立服务器域，可以理解为一个 mini 版的 RabbitMQ 服务器Broker：消息队列服务器实体

组件协同工作的执行流程如下：

消息生产者连接到 RabbitMQ Broker，创建 connection，开启 channel生产者声明交换机类型、名称、是否持久化等生产者发送消息，并指定消息是否持久化等属性，指定 routing keyexchange 收到消息后，根据 routing key 将消息路由到跟当前交换机绑定的相匹配的队列消费者监听消息队列，接收到消息后开始业务处理

从上述流程我们可以看到，消息首先要经过 Exchange 路由才能找到对应的 Queue。Exchange 有四种类型，分别是 Direct、Fanout、Topic、Headers

1. Direct Exchange

直连交换机，是一种带路由功能的交换机，需要绑定一个队列，绑定时要指定一个 RoutingKey（路由键）。生产者把消息发送到交换机时，也必须指定消息的 RoutingKey（路由键）。 Exchange 根据消息的 RoutingKey 进行判断，只有队列的 RoutingKey 与消息的 RoutingKey 一致，才会接收到消息

2. Fanout Exchange

扇形交换机，可以有多个消费者，每个消费者有自己的队列，每个队列都要绑定到交换机。生产者把消息发送到交换机，交换机把消息发给绑定过的所有队列，队列的消费者都能拿到消息，实现一条消息被多个消费者消费

3. Topic Exchange

主题交换机和直连交换机类似，也可以根据 RoutingKey 把消息路由到不同的队列，只不过 Topic 类型的交换机可以让队列在绑定 RoutingKey 时使用通配符。这种模型的 RoutingKey 一般由一个或多个单词组成，多个单词以 . 符号分割

通配符有两种：

* 符号：匹配一个词，比如 a.* 可以匹配 a.b、a.c，但是匹配不了 a.b.c# 符号：匹配一个或多个词，比如 rabbit.# 可以匹配 rabbit.a.b、rabbit.a，也可以匹配 rabbit.a.b.c

4. Headers Exchange

头部交换机不是用 RoutingKey 进行路由匹配，而是匹配请求头中所带的键值进行路由。创建队列需要设置绑定的头部信息，有两种模式：全部匹配和部分匹配，交换机会根据生产者发送过来的头部信息携带的键值去匹配队列绑定的键值，路由到对应的队列

SpringBoot 整合 RabbitMQ

生产者端引入依赖

org.springframework.boot

spring-boot-starter-amqp

生产者端添加配置

spring:

rabbitmq:

host: 127.0.0.1

port: 5672

username: root

password: 123

生产者端配置队列、交换机

@Configuration

public class RabbitMqConfig {

@Bean

public Queue rabbitmqTestDirectQueue() {

// Direct 队列

// name：队列名称

// durable：是否持久化

// exclusive：是否独享，如果设置 true，则只有创建者可以使用此队列

// autoDelete: 是否自动删除，也就是临时队列，当最后一个消费者断开连接就会自动删除

return new Queue("test_direct_queue", true, false, false);

}

@Bean

public DirectExchange rabbitmqTestDirectExchange() {

// Direct 交换机

return new DirectExchange("test_direct_exchange", true, false);

}

@Bean

public Binding bindDirect() {

return BindingBuilder

// // 绑定 Direct 队列

.bind(rabbitmqTestDirectQueue())

//到 Direct 交换机

.to(rabbitmqTestDirectExchange())

// 并设置匹配键

.with("test_direct_routing");

}

@Bean

public Queue rabbitmqTestFanoutQueueA() {

// fanout 队列 a

return new Queue("test_fanout_queue_a", true, false, false);

}

@Bean

public Queue rabbitmqTestFanoutQueueB() {

// fanout 队列 b

return new Queue("test_fanout_queue_b", true, false, false);

}

@Bean

public FanoutExchange rabbitmqTestFanoutExchange() {

// Fanout 交换机

return new FanoutExchange("test_fanout_exchange", true, false);

}

@Bean

public Binding bindFanoutA() {

return BindingBuilder

// 绑定 Fanout 队列 a

.bind(rabbitmqTestFanoutQueueA())

//到 Fanout 交换机

.to(rabbitmqTestFanoutExchange());

}

@Bean

public Binding bindFanoutB() {

return BindingBuilder

// 绑定 Fanout 队列 b

.bind(rabbitmqTestFanoutQueueB())

//到 Fanout 交换机

.to(rabbitmqTestFanoutExchange());

}

@Bean

public Queue rabbitmqTestTopicQueueA() {

// topic 队列 a

return new Queue("test_topic_queue_a", true, false, false);

}

@Bean

public Queue rabbitmqTestTopicQueueB() {

// topic 队列 b

return new Queue("test_topic_queue_b", true, false, false);

}

@Bean

public TopicExchange rabbitmqTestTopicExchange() {

// Topic 交换机

return new TopicExchange("test_topic_exchange", true, false);

}

@Bean

public Binding bindTopicA() {

return BindingBuilder

// 绑定 Topic 队列 a

.bind(rabbitmqTestTopicQueueA())

//到 Topic 交换机

.to(rabbitmqTestTopicExchange())

// 并设置匹配键

.with("a.*");

}

@Bean

public Binding bindTopicB() {

return BindingBuilder

// 绑定 Topic 队列 b

.bind(rabbitmqTestTopicQueueB())

//到 Topic 交换机

.to(rabbitmqTestTopicExchange())

// 并设置匹配键

.with("b.*");

}

@Bean

public Queue rabbitmqTestHeadersQueueA() {

// headers 队列 a

return new Queue("test_headers_queue_a", true, false, false);

}

@Bean

public Queue rabbitmqTestHeadersQueueB() {

// headers 队列 b

return new Queue("test_headers_queue_b", true, false, false);

}

@Bean

public HeadersExchange rabbitmqTestHeadersExchange() {

// Headers 交换机

return new HeadersExchange("test_headers_exchange", true, false);

}

@Bean

public Binding bindHeadersA() {

Map map = new HashMap<>();

map.put("key_a1", "a1");

map.put("key_a2", "a2");

return BindingBuilder

// 绑定 Headers 队列 a

.bind(rabbitmqTestHeadersQueueA())

//到 Headers 交换机

.to(rabbitmqTestHeadersExchange())

// 全部匹配

.whereAll(map).match();

}

@Bean

public Binding bindHeadersB() {

Map map = new HashMap<>();

map.put("key_b1", "b1");

map.put("key_b2", "b2");

return BindingBuilder

// 绑定 Headers 队列 b

.bind(rabbitmqTestHeadersQueueB())

//到 Headers 交换机

.to(rabbitmqTestHeadersExchange())

// 部分匹配

.whereAny(map).match();

}

}

生产者端发送消息

@Controller

public class RabbitMQController {

@Autowired

private RabbitTemplate rabbitTemplate;

@GetMapping("sendDirectMsg")

public String sendDirectMsg() {

Map map = new HashMap<>();

map.put("msg", "test_send_direct_msg");

rabbitTemplate.convertAndSend("test_direct_exchange", "test_direct_routing", map);

return "OK";

}

@GetMapping("sendFanoutMsg")

public String sendFanoutMsg() {

Map map = new HashMap<>();

map.put("msg", "test_send_fanout_msg");

rabbitTemplate.convertAndSend("test_fanout_exchange", "", map);

return "OK";

}

@GetMapping("sendTopicMsgA")

public String sendTopicMsgA() {

Map map = new HashMap<>();

map.put("msg", "test_send_topic_msg_a");

rabbitTemplate.convertAndSend("test_topic_exchange", "a.c", map);

return "OK";

}

@GetMapping("sendTopicMsgB")

public String sendTopicMsgB() {

Map map = new HashMap<>();

map.put("msg", "test_send_topic_msg_b");

rabbitTemplate.convertAndSend("test_topic_exchange", "b.c", map);

return "OK";

}

@GetMapping("sendHeadersMsgA")

public String sendHeadersMsgA() {

MessageProperties msgp = new MessageProperties();

msgp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);

msgp.setContentType("UTF-8");

Map map = new HashMap<>();

map.put("key_a1", "a1");

map.put("key_a2", "a2");

msgp.getHeaders().putAll(map);

Message msg = new Message("test_send_headers_msg_a".getBytes(), msgp);

rabbitTemplate.convertAndSend("test_headers_exchange", null, msg);

return "OK";

}

@GetMapping("sendHeadersMsgB")

public String sendHeadersMsgB() {

MessageProperties msgp = new MessageProperties();

msgp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);

msgp.setContentType("UTF-8");

Map map = new HashMap<>();

map.put("key_b1", "a1");

map.put("key_b3", "b3");

msgp.getHeaders().putAll(map);

Message msg = new Message("test_send_headers_msg_b".getBytes(), msgp);

rabbitTemplate.convertAndSend("test_headers_exchange", null, msg);

return "OK";

}

}

消费端引入的依赖和配置与生产端一样，接收消息

/**

* Direct 队列消费者

*/

@Slf4j

@Component

@RabbitListener(queues = {"test_direct_queue"})

public class RabbitMqDirectReceiver {

@RabbitHandler

public void process(Map map) {

log.info("消费者接收到消息: {}", map.toString());

}

}

/**

* Fanout 队列消费者 a

*/

@Slf4j

@Component

@RabbitListener(queues = {"test_fanout_queue_a"})

public class RabbitMqFanoutAReceiver {

@RabbitHandler

public void process(Map map) {

log.info("消费者接收到消息: {}", map.toString());

}

}

/**

* Fanout 队列消费者 b

*/

@Slf4j

@Component

@RabbitListener(queues = {"test_fanout_queue_b"})

public class RabbitMqFanoutBReceiver {

@RabbitHandler

public void process(Map map) {

log.info("消费者接收到消息: {}", map.toString());

}

}

/**

* Topic 队列消费者 a

*/

@Slf4j

@Component

@RabbitListener(queues = {"test_topic_queue_a"})

public class RabbitMqTopicAReceiver {

@RabbitHandler

public void process(Map map) {

log.info("消费者接收到消息: {}", map.toString());

}

}

/**

* Topic 队列消费者 b

*/

@Slf4j

@Component

@RabbitListener(queues = {"test_topic_queue_b"})

public class RabbitMqTopicBReceiver {

@RabbitHandler

public void process(Map map) {

log.info("消费者接收到消息: {}", map.toString());

}

}

/**

* Topic 队列消费者 a

*/

@Slf4j

@Component

@RabbitListener(queues = {"test_topic_queue_a"})

public class RabbitMqTopicAReceiver {

@RabbitHandler

public void process(Map map) {

log.info("消费者接收到消息: {}", map.toString());

}

}

/**

* Headers 队列消费者 a

*/

@Slf4j

@Component

public class RabbitMqHeadersAReceiver {

@RabbitListener(queuesToDeclare = @Queue("test_headers_queue_a"))

public void process(Message msg) throws Exception {

MessageProperties msgp = message.getMessageProperties();

String contentType = msgp.getContentType();

log.info("消费者接收到消息: {}", new String(message.getBody(), contentType));

}

}

/**

* Headers 队列消费者 b

*/

@Slf4j

@Component

public class RabbitMqHeadersBReceiver {

@RabbitListener(queuesToDeclare = @Queue("test_headers_queue_b"))

public void process(Message msg) throws Exception {

MessageProperties msgp = message.getMessageProperties();

String contentType = msgp.getContentType();

log.info("消费者接收到消息: {}", new String(message.getBody(), contentType));

}

}

RabbitMQ 推拉模型

RabbitMQ 有两种消息处理模式：推模式和拉模式

推模式下，生产者发布消息到队列时，会立即将这条消息发送给所有订阅该队列的消费者，优点：实现实时通信，缺点：如果消费者的处理能力跟不上生产者的速度，就会在消费者处造成消息堆积，因此需要根据消费能力做流控（比如 RabbitMQ 用 QOS 来限制），RabbitMQ 默认使用推消息

拉模式下，生产者发布消息到队列时，不会立即发送消息给消费者，而是等待消费者请求消息后才发送，优点：消费端可以按照自己的处理速度来消费，缺点：消息传递存在延迟，当处理速度小于发布速度时，容易造成消息堆积在队列

SpringBoot 实现拉消息代码如下：

@Slf4j

@Component

public class RabbitMQPullConsumer {

@Autowired

private RabbitTemplate rabbitTemplate;

public void process() {

rabbitTemplate.execute(new ChannelCallback