RabbitMQ的5种模式——再探RabbitMQ的模式，简单、工作，发布订阅_rabbitmq 订阅模式

先自我介绍一下，小编浙江大学毕业，去过华为、字节跳动等大厂，目前阿里P7

深知大多数程序员，想要提升技能，往往是自己摸索成长，但自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年最新软件测试全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上软件测试知识点，真正体系化！

由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新

如果你需要这些资料，可以添加V获取：vip1024b （备注软件测试）

正文

库存有没有可能挂了。或者访问量巨大。因为库存慢，导致订单也慢。

解耦。

3削峰填谷

双12的时候，叫好早上的8点。秒杀的时候。

有些瞬间服务器压力是超大的，过了这个瞬间，几乎没有消耗量。

等服务器能够正常的时候，我慢慢执行就行了。

常见的MQ

MQ的前身，就是一个发布者订阅模式。

Kafka RabbitMQ（1W/s） RocketMQ ActiveMQ

Kafka : 10w/S 主要用于日志。

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。

拆解控制台页面

总览页面

包括刷新时间间隔，rabbitmq节点，连接端口的信息等

配置可以导出和导入

连接connection

队列页面

包括队列的名字，状态，准备好的消息数量，未确认的消息数量，总计消息数量

简单模式

使用的依赖

com.rabbitmq amqp-client 5.7.3

工具类：建立连接

package com.tianju.config;

import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
* 建立连接的工具类
*/
public class ConnectionFactory {

public static Connection createConnection() throws IOException, TimeoutException {
com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();
connectionFactory.setHost(“192.168.111.130”); // http://192.168.111.130/
connectionFactory.setPort(5672);
connectionFactory.setUsername(“admin”);
connectionFactory.setPassword(“123”);
connectionFactory.setVirtualHost(“/demo”);

// amqp://admin@192.168.111.130:5672/
return connectionFactory.newConnection();
}
}

生产者：生产消息

package com.tianju.simple;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.tianju.config.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
* 生产者发送消息
* 建立连接–> 创建频道 --> 创建队列 --> 发送消息
*/
public class Provider {
private static String QUEUE_ORDER = “queue_order”;
public static void main(String[] args) throws IOException, TimeoutException {
// 建立连接
Connection connection = ConnectionFactory.createConnection();
// 创建频道
Channel channel = connection.createChannel();
// 创建队列
channel.queueDeclare(QUEUE_ORDER,false,false,false,null);

// 发送消息，指定给哪个队列上发消息
for (int i = 0; i < 100; i++) {
String msg = “hello rabbitmq–”+i;
channel.basicPublish(“”, QUEUE_ORDER, null, msg.getBytes());

System.out.println(“消息发布成功”);

}
connection.close();
}
}

消费者：消费消息

package com.tianju.simple;

import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {
private static String QUEUE_ORDER = “queue_order”;
public static void main(String[] args) throws IOException, TimeoutException {
// 建立连接
Connection connection = ConnectionFactory.createConnection();
// 创建频道
Channel channel = connection.createChannel();
// 创建队列
channel.queueDeclare(QUEUE_ORDER,false,false,false,null);
// 队列必须声明，如果不存在，则自动创建

// 声明一个消费者
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
/**
* 回调函数，用来接收消息
* @param consumerTag the consumer tag associated with the consumer
* @param envelope packaging data for the message
* @param properties content header data for the message
* @param body the message body (opaque, client-specific byte array)，字节数组
* @throws IOException
*/
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException
{
// no work to do

String msg = new String(body); // 收到的信息
System.out.println(“消费者接收到：”+msg);
try {
Thread.sleep(3000);// 模拟一个耗时操作
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
};
// 表明自己是消费者，接收消息
/**
* autoAck：自动确认设置成 true
*/
channel.basicConsume(QUEUE_ORDER, true, defaultConsumer);
}
}

进行测试

生产者发送100条消息

消费者进行消费，假设突然之间服务宕机了，此时消费了4条消息，理论上还应该有96条消息

打开控制台页面查看，消息全部消失了，出现了数据丢失的情况
！

全部用默认值的情况下，如果发生异常，则消息全部丢失。

消费者一次性拿到了所有的消息。

Ack：项目中必须false

生产者 ==投递消息=》队列

消费者=接受==》队列

ack: false 必须要手工确认。

消费者接到这个消息的时候， 这个消息进入 unack状态。

1：手工确认。消息删除。完成

2: 没手工确认，断开连接或者超时。

3：这个消息重新进入ready状态。等待其他消费者进行消费。

先设置自动ack为false，表示需要手工确认，然后在消费消息的方法中，进行消息的确认。

package com.tianju.simple;

import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {
private static String QUEUE_ORDER = “queue_order”;
public static void main(String[] args) throws IOException, TimeoutException {
// 建立连接
Connection connection = ConnectionFactory.createConnection();
// 创建频道
Channel channel = connection.createChannel();
// 创建队列
channel.queueDeclare(QUEUE_ORDER,false,false,false,null);
// 队列必须声明，如果不存在，则自动创建

// 声明一个消费者
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
/**
* 回调函数，用来接收消息
* @param consumerTag the consumer tag associated with the consumer
* @param envelope packaging data for the message
* @param properties content header data for the message
* @param body the message body (opaque, client-specific byte array)，字节数组
* @throws IOException
*/
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException {
// no work to do

String msg = new String(body); // 收到的信息
System.out.println(“消费者接收到：”+msg);
try {
Thread.sleep(3000);// 模拟一个耗时操作
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
// 消费者代码可能有失败，消息拿到之后，可能还没有处理，就宕机了
// 消息确认的代码一定在最后一行
// long deliveryTag 消息的下标；
// boolean multiple 是否批量确认；
channel.basicAck(envelope.getDeliveryTag(),true); // 确认消息，批量确认
}
};
// 表明自己是消费者，接收消息
/**
* autoAck：自动确认设置成 true
* 是否自动确认，
* false：不进行自动确认；true：自动确认
* 消费过程中可能产生异常
* 如果产生异常，则必须进行消费补偿
*/
channel.basicConsume(QUEUE_ORDER, false, defaultConsumer); // 接收消息
}
}

再次模拟宕机的情况，一开始消息全部被放到Unack中，当宕机时，又把消息吐了出来，至少消息没有出现丢失的情况。

工作模式

一个生产者，两个消费者

两个消费者消费的是同一个队列中的消息。

两个消费者上来后，默认是按照平均分配。结果： 有人瞬间干完。有人很久都没干完。

QOS: 限流

可以限制你一次拉取几个。

这样两个消费者，也能够节省服务器CPU了。

// 创建频道
Channel channel = connection.createChannel();
channel.basicQos(1); /** 1次只能拿1个 **/

当消费的速度太慢，觉得不够，加入其他的消费者。

核心，只有一个消息队列。

消息是轮询的方式，发送给不同的消费者。

Tips：队列参数怎么变？

队列的参数发生变化后，要删除再添加。

// 创建频道
Channel channel = connection.createChannel();

channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列

boolean durable = true; // 当前队列中的消息持久化操作，重启之后，消息还在
// 创建队列
channel.queueDeclare(QUEUE_ORDER,durable,false,false,null);

队列相关参数

x-max-length

package com.tianju.work;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.tianju.config.ConnectionFactory;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
* 生产者发送消息
* 建立连接–> 创建频道 --> 创建队列 --> 发送消息
*/
public class Provider {
private static String QUEUE_ORDER = “queue_order”;
public static void main(String[] args) throws IOException, TimeoutException {
// 建立连接
Connection connection = ConnectionFactory.createConnection();
// 创建频道
Channel channel = connection.createChannel();
channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列
boolean durable = true; // 当前队列中的消息持久化操作，重启之后，消息还在

Map map = new HashMap();
map.put(“x-max-length”, 10); // 设置最大的长度，只接受10个
// 创建队列
channel.queueDeclare(QUEUE_ORDER,durable,false,false,map);

// 发送消息，指定给哪个队列上发消息
for (int i = 0; i < 1000; i++) {
String msg = “hello rabbitmq–”+i;
channel.basicPublish(“”, QUEUE_ORDER, null, msg.getBytes());

System.out.println(“消息发布成功”);
}
connection.close();
}
}

How many (ready) messages a queue can contain before it starts to drop them from its head.

x-overflow

Sets the queue overflow behaviour. This determines what happens to messages when the maximum length of a queue is reached. Valid values are drop-head, reject-publish or reject-publish-dlx. The quorum queue type only supports drop-head and reject-publish.

此时就是前10个了

x-max-length-bytes

package com.tianju.work;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.tianju.config.ConnectionFactory;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
* 生产者发送消息
* 建立连接–> 创建频道 --> 创建队列 --> 发送消息
*/
public class Provider {
private static String QUEUE_ORDER = “queue_order”;
public static void main(String[] args) throws IOException, TimeoutException {
// 建立连接
Connection connection = ConnectionFactory.createConnection();
// 创建频道
Channel channel = connection.createChannel();
channel.queueDelete(QUEUE_ORDER); // 先删除旧的队列
boolean durable = true; // 当前队列中的消息持久化操作，重启之后，消息还在

Map map = new HashMap();
map.put(“x-max-length”, 10); // 设置最大的长度，只接受10个
map.put(“x-overflow”, “reject-publish”); // 拒绝发布，变成前10ge
map.put(“x-max-length-bytes”, 4); // 消息的最大字节长度
// 创建队列
channel.queueDeclare(QUEUE_ORDER,durable,false,false,map);

// 发送消息，指定给哪个队列上发消息
for (int i = 0; i < 20; i++) {
String msg = “53”+i;
channel.basicPublish(“”, QUEUE_ORDER, null, msg.getBytes());
msg = “hello rabbitmq–”+i;
channel.basicPublish(“”, QUEUE_ORDER, null, msg.getBytes());

System.out.println(“消息发布成功”);
}
connection.close();
}
}

参数

//创建队列
boolean durable = true;//当前队列中的消息进行持久化操作 重启之后，消息还在
Map map = new HashMap();
map.put(“x-max-length”,10);//设置队列的最大长度
map.put(“x-overflow”,“reject-publish”);//设置队列的最大长度 drop head
map.put(“x-max-length-bytes”,4);//设置消息的最大字节长度
// map.put(“x-expires”,10000);//超时后，直接删除队列
channel.queueDeclare(QUQUENAME,durable,false,false,map);

发布者订阅模式

工作模式中，只有一个队列。两个消费者轮询从同一个队列中取数据。

发布者订阅模式： 交换机后面会绑定多个消息队列。每个消息队列都有完整的信息。每个队列后的消费者都会有完整的消息。

交换机：就是一个无意识的广播。行为就是扇出.

场景：

下订单：

1：订单的数据入数据库
2：发送订单的短信
3： 物流

要每个队列都有完整的信息。

生产者：给fanout交换机发消息

package com.tianju.publish;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.MessageProperties;
import com.tianju.config.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
* 生产者发送消息
* 建立连接–> 创建频道 --> 创建队列 --> 发送消息
*/
public class Provider {
private static String QUEUE_ORDER = “queue_order”;
private static String EXCHANGE = “pet_exchange”;
public static void main(String[] args) throws IOException, TimeoutException {
// 建立连接
Connection connection = ConnectionFactory.createConnection();
// 创建频道
Channel channel = connection.createChannel();

// 创建交换机
channel.exchangeDeclare(EXCHANGE, “fanout”); // 扇出，类型只能用这个

for (int i = 0; i < 100; i++) {
channel.basicPublish(EXCHANGE, “”, MessageProperties.TEXT_PLAIN, (“hello fanout”+i).getBytes());
}
}
}

消费者1：队列q32

package com.tianju.publish;

import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {
private static String EXCHANGE = “pet_exchange”;

public static void main(String[] args) throws IOException, TimeoutException {
// 建立连接
Connection connection = ConnectionFactory.createConnection();
// 创建频道
Channel channel = connection.createChannel();

// 创建交换机
channel.exchangeDeclare(EXCHANGE, “fanout”); // 扇出，类型只能用这个

channel.queueDeclare(“q32”, false, false, false, null);
channel.queueBind(“q32”, EXCHANGE, “”); // 路由键fanout模式，必须为空，即使写了是无效的

channel.basicConsume(“q32”, new DefaultConsumer(channel){
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException
{
// no work to do
String msg = new String(body);
System.out.println(“消费者1：”+msg);
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}

消费者2：队列q321

package com.tianju.publish;

import com.rabbitmq.client.*;
import com.tianju.config.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer2 {
private static String EXCHANGE = “pet_exchange”;

public static void main(String[] args) throws IOException, TimeoutException {
// 建立连接
Connection connection = ConnectionFactory.createConnection();
// 创建频道
Channel channel = connection.createChannel();

// 创建交换机
channel.exchangeDeclare(EXCHANGE, “fanout”); // 扇出，类型只能用这个

channel.queueDeclare(“q321”, false, false, false, null);
channel.queueBind(“q321”, EXCHANGE, “”); // 路由键fanout模式，必须为空，即使写了是无效的

channel.basicConsume(“q321”, new DefaultConsumer(channel){
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException
{
// no work to do
String msg = new String(body);
System.out.println(“消费者2：”+msg);
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化的资料的朋友，可以添加V获取：vip1024b （备注软件测试）

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！
l();

// 创建交换机
channel.exchangeDeclare(EXCHANGE, “fanout”); // 扇出，类型只能用这个

channel.queueDeclare(“q321”, false, false, false, null);
channel.queueBind(“q321”, EXCHANGE, “”); // 路由键fanout模式，必须为空，即使写了是无效的

channel.basicConsume(“q321”, new DefaultConsumer(channel){
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException
{
// no work to do
String msg = new String(body);
System.out.println(“消费者2：”+msg);
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化的资料的朋友，可以添加V获取：vip1024b （备注软件测试）
[外链图片转存中…(img-L1jPePzA-1713357665560)]

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！