您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生
尚硅谷RabbitMQ视频-028-进阶篇-消费端限流
发布时间：2026-11-06 22:51:01编辑：雪饮阅读（）
-
这次直接建立新的模块
D:\rabbitMqLearning\currentLimiting\pom.xml:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>rabbitMqLearning</artifactId>
<groupId>org.example</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>currentLimiting</artifactId>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.1.5</version>
</parent>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.16.0</version>
</dependency>
<!-- spring-boot rabbitmq amqp 依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- spring-boot web依赖 -->
<!-- (注意默认会占用本地8080端口，若已经被占用，则可能无法启动) -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- spring-boot lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<!--用测试依赖来实现实现生产者-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
</dependencies>
</project>
D:\rabbitMqLearning\currentLimiting\src\main\resources\application.yml:
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: 123456
virtual-host: /
listener:
simple:
acknowledge-mode: manual # 把消息确认模式改为手动确认,因为默认都是按成功处理，默认都是ack
publisher-confirm-type: correlated #交换机的确认
publisher-returns: true #队列的确认D:\rabbitMqLearning\currentLimiting\src\main\java\org\example\RabbitMQConsumerMainType.java:
package org.example;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class RabbitMQConsumerMainType {
public static void main(String[] args) {
SpringApplication.run(RabbitMQConsumerMainType.class, args);
}
}D:\rabbitMqLearning\currentLimiting\src\main\java\org\example\listener\MyMessageListener.java:
package org.example.listener;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
@Slf4j
public class MyMessageListener {
public static final String QUEUE_NAME = "queue.order";
@RabbitListener(queues = {QUEUE_NAME})
public void processMessage(String dataString, Message message, Channel channel) throws IOException {
long deliveryTag=message.getMessageProperties().getDeliveryTag();
try {
// 核心操作 处理消息
log.info("org.example.confirmConsumer.listener 消费端 消息内容: " + dataString);
/*
核心操作成功: 返回 ACK 信息，
multiple参数为：true为批量操作，false为非批量操作,
basicAck作用：**成功确认**，Broker 收到 ack 之后，直接删除这条消息
*/
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
/*
获取当前消息是否是重复投递的
redelivered 为 true：说明当前消息已经重复投递过一次了
redelivered 为 false：说明当前消息是第一次投递
*/
Boolean redelivered = message.getMessageProperties().getRedelivered();
/*
核心操作失败: 返回 NACK 信息
requeue 参数: 控制消息是否重新放回队列
取值为 true: 重新放回队列, broker 会重新投递这个消息
取值为 false: 不重新放回队列, broker 会丢弃这个消息
basicNack作用：**失败拒绝**
*/
//已是重投消息，则拒绝重投
if(redelivered){
channel.basicNack(deliveryTag, false, false);
}
//非重投消息，则重新投递
if(!redelivered){
channel.basicNack(deliveryTag, false, true);
}
}
}
}接下来就是生产者先批量发一堆消息D:\rabbitMqLearning\currentLimiting\src\test\java\org\example\RabbitMQTest.java:
package org.example;
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;
@SpringBootTest
public class RabbitMQTest {
public static final String EXCHANGE_DIRECT = "exchange.direct.order";
public static final String ROUTING_KEY = "order";
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void test02SendMessage() {
for (int i = 0; i < 100; i++) {
rabbitTemplate.convertAndSend(EXCHANGE_DIRECT, ROUTING_KEY, "Test Prefetch " + i);
}
}
}现在执行test02SendMessage（模拟产生消息积压）后100个消息也秒消费了。然后修改D:\rabbitMqLearning\currentLimiting\src\main\java\org\example\listener\MyMessageListener.java:
package org.example.listener;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
public class MyMessageListener {
public static final String QUEUE_NAME = "queue.order";
//@RabbitListener(queues = {QUEUE_NAME})
public void processMessage(String dataString, Message message, Channel channel) throws IOException {
long deliveryTag=message.getMessageProperties().getDeliveryTag();
try {
// 核心操作 处理消息
log.info("org.example.confirmConsumer.listener 消费端 消息内容: " + dataString);
/*
核心操作成功: 返回 ACK 信息，
multiple参数为：true为批量操作，false为非批量操作,
basicAck作用：**成功确认**，Broker 收到 ack 之后，直接删除这条消息
*/
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
/*
获取当前消息是否是重复投递的
redelivered 为 true：说明当前消息已经重复投递过一次了
redelivered 为 false：说明当前消息是第一次投递
*/
Boolean redelivered = message.getMessageProperties().getRedelivered();
/*
核心操作失败: 返回 NACK 信息
requeue 参数: 控制消息是否重新放回队列
取值为 true: 重新放回队列, broker 会重新投递这个消息
取值为 false: 不重新放回队列, broker 会丢弃这个消息
basicNack作用：**失败拒绝**
*/
//已是重投消息，则拒绝重投
if(redelivered){
channel.basicNack(deliveryTag, false, false);
}
//非重投消息，则重新投递
if(!redelivered){
channel.basicNack(deliveryTag, false, true);
}
}
}
@RabbitListener(queues = {QUEUE_NAME})
public void processMessageTestPrefetch(String dataString, Message message, Channel channel) throws IOException, InterruptedException {
log.info("消费端 消息内容: " + dataString);
//这里是模拟实际业务逻辑，因为实际业务逻辑处理一般情况是会比较慢一点
TimeUnit.SECONDS.sleep(1);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}
注释了原来哪个秒处理的processMessage的注解，所以processMessage
不会被调用，没有注解不被扫描到，然后processMessage也没有被主动调用的地方。
然后重新运行RabbitMQConsumerMainType.main方法及test02SendMessage
方法就1秒一个一个的来消费处理每个消息都睡一下
发现取的快，但确认是慢慢的确认（ready为0，unacked和total是慢慢向下递减），在rabbitmq管理ui中的队列queue.order中能够看到。
prefetch = N：每个消费者（每个 channel）身上最多同时挂着 N 条已经送达、但还没回 ack 的消息。
差别在哪：
- 消息被 broker 推给消费者的那一刻，它就已经算 "在你手上了"
- 不管你是还没开始处理，还是正在处理中，还是处理完忘了 ack—— 只要没调用 basicAck，它都占着这个名额
- 名额满了（N 条全占着），broker 就停推
- 某一条 ack 了，空出一个位置，broker 才补推下一条
prefetch = 一个骑手手机上同时能 "挂" 几单没送完的
application.yml文件中的配置类比：
spring.rabbitmq.host: localhost # 连的是"美团平台服务器"（broker） port: 5672 # 平台大门端口
username: guest # 账号
application.yml文件修改后内容：
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: 123456
virtual-host: /
listener:
simple:
acknowledge-mode: manual # 把消息确认模式改为手动确认,因为默认都是按成功处理，默认都是ack
prefetch: 1 # 限流闸门：每个消费者最多同时持有 1 条未确认(unacked)消息；处理完并手动 ack 后，broker 才推下一条，防止消息洪流瞬间压垮消费者
publisher-confirm-type: correlated #交换机的确认
publisher-returns: true #队列的确认关键字词：RabbitMQ,视频,进阶篇,消费端,限流
上一篇：尚硅谷RabbitMQ视频-027-进阶篇-可靠性-消端确-测试
下一篇：返回列表