RabbitMQ高级特性（三）：RabbitMQ实现消费端限流Qos
一、生产者工程
（1）RabbitMQ配置文件（rabbitmq.properties）
rabbitmq.host=192.168.116.161
rabbitmq.port=5672
rabbitmq.username=xiao
rabbitmq.password=xiao
rabbitmq.virtual-host=/myhost
（2）声明队列、交换机、交换机绑定队列（spring-rabbitmq-producer.xml）
<!--消息可靠性投递（生产端）-->
<!-- （1）队列 -->
<rabbit:queue id="test_queue_confirm" name="test_queue_confirm"/>
<!-- （2）交换机：绑定队列 -->
<rabbit:direct-exchange name="test_exchange_confirm">
<rabbit:bindings>
<rabbit:binding queue="test_queue_confirm" key="confirm"/>
</rabbit:bindings>
</rabbit:direct-exchange>
（2）测试类（发送消息到队列中）
package net.xiaof.test;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 测试消费端限流（只负责发送数据）
*/
@Test
public void testQos() {
//发送10条消息
for (int i = 0; i < 10; i++) {
// 发送消息
rabbitTemplate.convertAndSend("test_exchange_confirm", "confirm", "（" + (i + 1) + "） message confirm....");
}
}
}
二、消费者工程
（1）RabbitMQ配置文件（rabbitmq.properties）
rabbitmq.host=192.168.116.161
rabbitmq.port=5672
rabbitmq.username=xiao
rabbitmq.password=xiao
rabbitmq.virtual-host=/myhost
（2）自定义限流监听器（注：为看效果，这里注释了签收代码）
package net.xiaof.listener;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
/**
* @author zhangxh
* @Description: 消费端限流（每次处理指定数量消息）
* @date 2020-12-26
*/
@Component
public class QosListener implements ChannelAwareMessageListener {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
System.out.println("QosListener的onMessage方法执行了");
//1.在监听器容器<rabbit:listener-container>中开启“手动确认模式”：acknowledge="manual"
//2.在监听器容器<rabbit:listener-container>中设置预处理数量（这里预处理2条）：prefetch="2"
try {
//3.重写onMessage方法，在onMessage方法中处理业务逻辑
System.out.println("处理业务逻辑...");
//4.消息处理成功：完成手动签收，channel.basicAck(...)
// channel.basicAck(deliveryTag,true);
} catch (Exception e) {
//e.printStackTrace();
Thread.sleep(1200);//防止重回队列与消费过程打印过快
//5.消息处理异常：在catch中进行拒绝签收，channel.basicNack(...)拒绝签收，消息重新回到queue，broker重新发送给consumer
/**
* long deliveryTag:该消息的index
* boolean multiple：是否批量。true:将一次性拒绝所有小于deliveryTag的消息。
* boolean requeue：被拒绝的是否重新入队列
*/
// channel.basicNack(deliveryTag,true,true);
}
}
}
（3）开启手动确认模式、设置预处理消息数量、声明Qos监听器（spring-rabbitmq-consumer.xml）
步骤：
① 在监听器容器<rabbit:listener-container>属性中开启“手动确认模式”：acknowledge=“manual”
② 在监听器容器<rabbit:listener-container>属性中设置预处理数量（这里预处理2条）：prefetch=“2”
③ 在监听器容器<rabbit:listener-container>的子标签声明Qos监听器
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
<!--加载配置文件-->
<context:property-placeholder location="classpath:rabbitmq.properties"/>
<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
port="${rabbitmq.port}"
username="${rabbitmq.username}"
password="${rabbitmq.password}"
virtual-host="${rabbitmq.virtual-host}"/>
<!-- 组件扫描 -->
<context:component-scan base-package="net.xiaof.listener" />
<!--定义监听器容器（设置手动签收模式：acknowledge="manual"，prefetch预处理数量）-->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual" prefetch="2" >
<!-- 自定义QOS监听器，监听队列test_queue_confirm -->
<rabbit:listener ref="qosListener" queue-names="test_queue_confirm"/>
</rabbit:listener-container>
</beans>
（4）测试类（运行死循环测试类）
package net.xiaof.test;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author zhangxh
* @Description: 测试
* @date 2020-12-25
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-consumer.xml")
public class ConsumerTest {
/**
* 使用死循环开启消费端
*/
@Test
public void testConsumer() throws InterruptedException {
while (true){
}
}
}
三、测试方法
（1）运行生产者工程测试类，发送消息到“test_queue_confirm”队列中。
（2）运行消费者工程测试类，消费指定数量的消息（在Qos监听器中设置类预处理2条消息）
注：这里在Qos监听器中注释了“手动签收”的代码（net.xiaof.listener.QosListener）。
//4.消息处理成功：完成手动签收，channel.basicAck(...)
// channel.basicAck(deliveryTag,true);
运行结果：
然后放开“手动签收”的代码（net.xiaof.listener.QosListener）：
//4.消息处理成功：完成手动签收，channel.basicAck(...)
channel.basicAck(deliveryTag,true);
重新运行消费者工程测试类，运行结果：
