文章目录
RabbitMQ 消息分发
概述
RabbitMQ 的队列在有多个消费者订阅时,默认会通过轮询
的机制将消息分发给不同的消费者,但是有些消费者消费速度慢,有些消费者消费速度快,就会导致消费速度慢的消费者影响整个的任务的吞吐量下降
.
例如,公司有1个正式员工和1个实习生,现在有 10 个任务分配平均给他们(各 5 个),而由于实习生干活比较慢,就会导致整个完成任务的吞吐量下降.
消息分发机制给 “正式工” 多分一些任务,给 “实习生” 少分一些任务.
如何实现消费分发机制(限制每个队列消息数量)
可以在配置文件中配置 prefetchCount
(或者使用原生的 channel.basicQos(int prefetchCount)
),来限制当前消息通道上(channel)的每一个消费所能保持的最大未确认消息的数量.
例如 prefetchCount 为 10,并且一个 channel 上有两个消费者,那么每个消费者都最多接收 10 条未确认的消息. 此时整个 channel 上未确认消息总数可能达到 20 条.
具体使用:例如配置 prefetch = 5
,那么 RabbitMQ 就会为消费者计数. 发送一条消息计数+1,消费一条消息计数-1,当达到了上限5,mq队列 就不会再发送消息,直到消费者确认了某条消息(类似 TCP 中的华滑动窗口).
使用场景
限流
背景
假设,订单系统每秒最多处理 1000 请求,正常情况下,该订单系统可以满足日常使用.
但是在突发的秒杀场景下,请求瞬间增多,每秒 1w qps,这不得把订单系统打成筛子.
问题:mq 在中间的话,不是已经有削峰填谷的作用了么?为什么还要使用 mq 的 prefetch 限流机制?
尽管消息队列可以延缓高峰压力,但消费者的处理能力还是有限的(如果不配置 prefetch,消费者自身从队列中取消息的量是不可控的)
. 如果消费者一次性取走过多的消息,就可能会导致资源紧张. prefetch 限流就是用来控制每个消费者取消息的数量,确保消费者不会过载.
实现 demo
假设限制未确认消息上限为 5,发送消息数量为 20.
a)配置 prefetch 参数,设置应答方式为手动应答.
spring:
application:
name: rabbitmq
rabbitmq:
host: env-base
port: 5672
username: root
password: 1111
listener:
simple:
acknowledge-mode: manual # 手动确认
prefetch: 5
b)配置交换机队列
@Configuration
class MQConfig {
@Bean
fun transQueue() = Queue(MQConst.TRANS_QUEUE)
@Bean
fun qosExchange() = DirectExchange(MQConst.QOS_EXCHANGE)
@Bean
fun qosQueue() = Queue(MQConst.QOS_QUEUE)
@Bean
fun qosBinding(): Binding = BindingBuilder
.bind(qosQueue())
.to(qosExchange())
.with(MQConst.QOS_BINDING_KEY)
}
c)接口(生产者)
@RestController
@RequestMapping("/mq")
class MQApi(
val rabbitTemplate: RabbitTemplate,
) {
@RequestMapping("/qos")
fun qos(): String {
for (i in 1..20) {
rabbitTemplate.convertAndSend(MQConst.QOS_EXCHANGE, MQConst.QOS_BINDING_KEY, "qos msg $i")
}
return "ok"
}
}
d)消费者
@Component
class QosListener {
@RabbitListener(queues = [MQConst.QOS_QUEUE])
fun handMessage(
message: Message,
channel: Channel
) {
val deliverTag = message.messageProperties.deliveryTag
try {
println("接收到消息: ${String(message.body, charset("UTF-8"))}, ${message.messageProperties.messageId}")
// 这里不主动应答,模拟超长业务
// channel.basicAck(deliverTag, false)
} catch (e: Exception) {
channel.basicNack(deliverTag, false, true)
}
}
}
e)效果如下:
可以观察到,消费者只接收到 5 个消息,但由于没有主动应答,队列 就不会给消费者发送新的消息.
Ps:此时如果直接关闭程序,这 5 个为应答的消息就会重回队列,成为 Ready 状态.
如下可以直接清理掉这些消息:
非公平发送(负载均衡)
背景
假设有两个消费者,mq 默认会按照轮询的策略将消息分发给消费者.
*但有一个中情况就比较尴尬:打个比方 一个是正式工,另一个是实习生,正式工就处理的很快,而实习生就很慢,就会造成整个任务的进度被拖慢. *
因此我们可以通过 负载均衡 的方式,让处理的快的消费者多处理一些,处理慢的消费者少处理一些.
具体的:只需要配置 prefetch,并开启自动应答即可.
这样一来,处理的快的消费者,自动应答的就更快,接收的消息也就更多.
实现 demo
a)配置文件
spring:
application:
name: rabbitmq
rabbitmq:
host: env-base
port: 5672
username: root
password: 1111
listener:
simple:
acknowledge-mode: manual # 手动确认
prefetch: 1 # 具体配置为多少,需要根据实际业务以及系统承受能力(压测)
b)生产者
@RequestMapping("/qos")
fun qos(): String {
for (i in 1..20) {
rabbitTemplate.convertAndSend(MQConst.QOS_EXCHANGE, MQConst.QOS_BINDING_KEY, "qos msg $i")
}
return "ok"
}
c)两个消费者
@Component
class QosListener {
@RabbitListener(queues = [MQConst.QOS_QUEUE])
fun fastHandMessage(
message: Message,
channel: Channel
) {
val deliverTag = message.messageProperties.deliveryTag
try {
println("接收到消息: ${String(message.body, charset("UTF-8"))}, ${message.messageProperties.messageId}")
Thread.sleep(1000)
println("正式工: 任务处理完成!")
channel.basicAck(deliverTag, false)
} catch (e: Exception) {
channel.basicNack(deliverTag, false, true)
}
}
@RabbitListener(queues = [MQConst.QOS_QUEUE])
fun slowHandMessage(
message: Message,
channel: Channel
) {
val deliverTag = message.messageProperties.deliveryTag
try {
println("接收到消息: ${String(message.body, charset("UTF-8"))}, ${message.messageProperties.messageId}")
Thread.sleep(2000)
println("实习生: 任务处理完成!")
channel.basicAck(deliverTag, false)
} catch (e: Exception) {
channel.basicNack(deliverTag, false, true)
}
}
}
d)效果如下: