RabbitMQ的5种模式——再探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://[email protected]: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行业感爱好的新人，都欢迎参加我们的的圈子（技能互换、学习资源、职场吐槽、大厂内推、口试辅导），让我们一起学习发展！