欢迎光临
我们一直在努力
        找回密码
当前位置：171主机测评 > 服务器技术 > 正文

RabbitMQ的五大常用模式

2026-10-09 分类：服务器技术 阅读(12) 评论(0)

        RabbitMQ 的“五大工作模式”是基于其核心组件（交换机、队列、绑定）的不同组合方式而形成的。前两种模式（简单、工作队列）实际上使用的是 RabbitMQ 内置的默认交换机，而后三种模式（发布/订阅、路由、主题）则需要显式地声明特定类型的交换机来实现灵活的消息分发

一.简单模式

  • 结构：一个生产者（P）将消息直接发送到队列，一个消费者（C）从该队列获取消息。

  • 核心特点：消息被消费后即刻从队列移除，一条消息只会被一个消费者接收。

        步骤：
        1.创建工程（生产者，消费者）
        2.分别添加依赖
        3.编写生产者发送消息
        4.编写消费者接收消息
生产者代码：

public class producer_HelloWorld {
public static void main(String[] args) throws IOException, TimeoutException {

//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
//（队列名称，是否持久化当mq重启后还在，是否独占，只能有一个消费者监听这个队列。是否自动删除，参数）
//如果没有一个名字是hello world的队列，则会创建该队列，如果有则不会创建
channel.queueDeclare("hello world",true,false,false,null);

//6.发送消息
/**
* exchange:交换机名称，简单模式下交换机会使用默认的“”
* routingkey:路由名称
* props:配置信息
* body:发送消息数据
*/
String body = "hello rabbitmq";
channel.basicPublish("","hello world",null,body.getBytes());
//7.释放资源 先关 channel，再关 connection（顺序反了会报错）
channel.close();
connection.close();
}
}
创建连接工厂—-设置参数—-创建连接Connection—-创建channel—-创建队列queue—-发送消息—-释放资源

消费者代码：

public class comsume_HelloWorld {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
//（队列名称，是否持久化当mq重启后还在，是否独占，只能有一个消费者监听这个队列。是否自动删除，参数）
//如果没有一个名字是hello world的队列，则会创建该队列，如果有则不会创建
channel.queueDeclare("hello world",true,false,false,null);
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println("consumerTag:"+consumerTag);
System.out.println("envelope:"+envelope.getExchange());
System.out.println("envelope:"+envelope.getRoutingKey());
System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
}
};
channel.basicConsume("hello world",true,comsumer);
}
}
创建连接工厂—-设置参数—-创建连接Connection—-创建channel—-创建队列queue—-接收消息

二.工作队列模式

  • 结构：一个生产者对应一个队列，但有多个消费者（C1, C2…）同时监听同一个队列。

  • 核心特点：消息被竞争消费（默认轮询分发，即 C1 拿一条，C2 拿一条）。这意味着同一条消息只能被其中一个消费者获取，主要用于负载均衡或处理耗时任务（如抢票、发短信）。

生产者代码：

public class producer_WorkQueues {
public static void main(String[] args) throws IOException, TimeoutException {

//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
//（队列名称，是否持久化当mq重启后还在，是否独占，只能有一个消费者监听这个队列。是否自动删除，参数）
//如果没有一个名字是hello world的队列，则会创建该队列，如果有则不会创建
channel.queueDeclare("work_queues",true,false,false,null);

//6.发送消息
/**
* exchange:交换机名称，简单模式下交换机会使用默认的“”
* routingkey:路由名称
* props:配置信息
* body:发送消息数据
*/
for (int i = 0; i < 10; i++) {
String body = i+"hello rabbitmq";
channel.basicPublish("","work_queues",null,body.getBytes());
}

//7.释放资源 先关 channel，再关 connection（顺序反了会报错）
channel.close();
connection.close();
}
}

消费者代码：

public class comsume_WorkQueues1 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
//（队列名称，是否持久化当mq重启后还在，是否独占，只能有一个消费者监听这个队列。是否自动删除，参数）
//如果没有一个名字是hello world的队列，则会创建该队列，如果有则不会创建
channel.queueDeclare("work_queues",true,false,false,null);
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag:"+consumerTag);
// System.out.println("envelope:"+envelope.getExchange());
// System.out.println("envelope:"+envelope.getRoutingKey());
// System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
}
};
channel.basicConsume("work_queues",true,comsumer);
}
}

public class comsume_WorkQueues2 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
//（队列名称，是否持久化当mq重启后还在，是否独占，只能有一个消费者监听这个队列。是否自动删除，参数）
//如果没有一个名字是hello world的队列，则会创建该队列，如果有则不会创建
channel.queueDeclare("work_queues",true,false,false,null);
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag:"+consumerTag);
// System.out.println("envelope:"+envelope.getExchange());
// System.out.println("envelope:"+envelope.getRoutingKey());
// System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
}
};
channel.basicConsume("work_queues",true,comsumer);
}
}
注：是多个消费者

三.Pub/Sub订阅发布模式

  • 结构：生产者将消息发送到 Fanout 交换机。该交换机会将消息无视路由键（Routing Key），广播给所有与之绑定的队列。每个队列对应一个消费者。

  • 核心特点：一条消息会被多个消费者同时接收。适用于“国家气象局发天气，百度/腾讯/网易同时接收”这种一对多的广播场景。

区别：在生产者和消息队列之间加了交换机，几种类型如下：

Fanout:广播，将消息交给所有绑定到交换机的队列

Direct:定向，把消息交给符合指定routing key的队列

Topic:通配符，把消息交给符合routing pattern（路由模式）的队列

生产者代码：

public class producer_PubSub {
public static void main(String[] args) throws IOException, TimeoutException {

//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();
//5.创建交换机
/**
* exchange:交换机名称
* type:交换机类型
* durable:是否持久化
* autoDelete:自动删除
* internal:内部使用，一般是false
* arguments:参数
*/
String exchangeName = "test_fanout";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);
//6.创建队列
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//7.绑定队列和交换机
/**
* queue:队列名称
* exchange:交换机名称
* routingKey:路由键，绑定规则
* 如果交换机的类型是fanout,routingKey设置为“”
*/
channel.queueBind(queue1Name,exchangeName,"");
channel.queueBind(queue2Name,exchangeName,"");
//8.发送消息
String body = "日志信息";
channel.basicPublish(exchangeName,"",null,body.getBytes());
//9.释放资源
channel.close();
connection.close();
}
}
相比较工作模式，这里多了创建交换机，绑定交换机和队列这两步

消费者代码：

public class comsume_PubSub1 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag:"+consumerTag);
// System.out.println("envelope:"+envelope.getExchange());
// System.out.println("envelope:"+envelope.getRoutingKey());
// System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
System.out.println("将日志打印；");
}
};
channel.basicConsume(queue1Name,true,comsumer);
}
}

public class comsume_PubSub2 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
String queue1Name = "test_fanout_queue1";
String queue2Name = "test_fanout_queue2";
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag:"+consumerTag);
// System.out.println("envelope:"+envelope.getExchange());
// System.out.println("envelope:"+envelope.getRoutingKey());
// System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
System.out.println("将日志打印保存数据库；");
}
};
channel.basicConsume(queue2Name,true,comsumer);
}
}
在这里这两个消费者都可以读到数据，不是轮询的

四.路由模式

  • 结构：生产者发送消息时指定一个 Routing Key。队列绑定到交换机时也指定一个或多个 Routing Key。只有当消息的 Routing Key 与队列的绑定 Key 完全匹配时，消息才会被路由到该队列。

  • 核心特点：实现了定向精准投递。例如，级别为 error 的日志只会发给专门处理错误的队列，而不会被信息队列接收。

注意：

1.队列与交换机绑定，不是任意绑定了，而是要所指定一个routingkey

2.消息的发送方在向exchange发送消息的时候，也必须指定routingkey

3.exchange不在把消息交给每一个绑定的队列，而是根据routingkey进行判断，只有队列的routingkey和消息的routingkey一样是，才会接收到消息

生产者代码：

public class producer_Routing {
public static void main(String[] args) throws IOException, TimeoutException {

//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();
//5.创建交换机
/**
* exchange:交换机名称
* type:交换机类型
* durable:是否持久化
* autoDelete:自动删除
* internal:内部使用，一般是false
* arguments:参数
*/
String exchangeName = "test_direct";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);
//6.创建队列
String queue1Name = "test_direct_queue1";
String queue2Name = "test_direct_queue2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//7.绑定队列和交换机
/**
* queue:队列名称
* exchange:交换机名称
* routingKey:路由键，绑定规则
* 如果交换机的类型是fanout,routingKey设置为“”
*/
channel.queueBind(queue1Name,exchangeName,"error");
channel.queueBind(queue2Name,exchangeName,"info");
channel.queueBind(queue2Name,exchangeName,"error");
channel.queueBind(queue2Name,exchangeName,"waring");
//8.发送消息
String body = "日志信息";
channel.basicPublish(exchangeName,"info",null,body.getBytes());

//9.释放资源
channel.close();
connection.close();
}
}
这里与发布订阅模式不同在，这个模式写明了routingkey的要求

消费者代码：

public class comsume_Rouring1 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
String queue1Name = "test_direct_queue1";
String queue2Name = "test_direct_queue2";
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag:"+consumerTag);
// System.out.println("envelope:"+envelope.getExchange());
// System.out.println("envelope:"+envelope.getRoutingKey());
// System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
System.out.println("将日志打印；");
}
};
channel.basicConsume(queue1Name,true,comsumer);
}
}

public class comsume_Rouring2 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
String queue1Name = "test_direct_queue1";
String queue2Name = "test_direct_queue2";
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag:"+consumerTag);
// System.out.println("envelope:"+envelope.getExchange());
// System.out.println("envelope:"+envelope.getRoutingKey());
// System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
System.out.println("将日志打印到控制台；");
}
};
channel.basicConsume(queue2Name,true,comsumer);
}
}
注意routingkey要与对应的生产者的routingkey相一致

五.主题模式

  • 结构：与路由模式类似，但 Routing Key 支持通配符：

    • * ：匹配一个单词。

    • # ：匹配零个或多个单词。

  • 核心特点：实现了灵活的一对多路由。例如，绑定 log.*.error 可以匹配 log.order.error 和 log.user.error；绑定 log.# 则可以匹配 log.info、log.order.error 等任意以 log 开头的消息。

生产者代码:

public class producer_Topic {
public static void main(String[] args) throws IOException, TimeoutException {

//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();
//5.创建交换机
/**
* exchange:交换机名称
* type:交换机类型
* durable:是否持久化
* autoDelete:自动删除
* internal:内部使用，一般是false
* arguments:参数
*/
String exchangeName = "test_topic";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);
//6.创建队列
String queue1Name = "test_topic1";
String queue2Name = "test_topic2";
channel.queueDeclare(queue1Name,true,false,false,null);
channel.queueDeclare(queue2Name,true,false,false,null);
//7.绑定队列和交换机
/**
* queue:队列名称
* exchange:交换机名称
* routingKey:路由键，绑定规则
* 如果交换机的类型是fanout,routingKey设置为“”
*/

channel.queueBind(queue1Name,exchangeName,"#.error");
channel.queueBind(queue1Name,exchangeName,"order.*");
channel.queueBind(queue2Name,exchangeName,"*.*");
//8.发送消息
String body = "日志信息";
channel.basicPublish(exchangeName,"order.info",null,body.getBytes());
//9.释放资源
channel.close();
connection.close();
}
}

消费者代码：

public class comsume_Topic1 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
String queue1Name = "test_topic1";
String queue2Name = "test_topic2";
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag:"+consumerTag);
// System.out.println("envelope:"+envelope.getExchange());
// System.out.println("envelope:"+envelope.getRoutingKey());
// System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
System.out.println("将日志打印；");
}
};
channel.basicConsume(queue1Name,true,comsumer);
}
}

public class comsume_Topic2 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
//2.设置参数
factory.setHost("127.0.0.1");//ip地址
factory.setPort(5672);
factory.setVirtualHost("/idcast");//虚拟机
factory.setUsername("yang");
factory.setPassword("yang");

//3.创建连接 Connection
Connection connection = factory.newConnection();
//4.创建channel
Channel channel = connection.createChannel();

//5.创建队列queue
String queue1Name = "test_topic1";
String queue2Name = "test_topic2";
//6.接收消息
/**
* queue:队列名称
* autoAck:是否自动确认
* callback:回调对象
*/
Consumer comsumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
// System.out.println("consumerTag:"+consumerTag);
// System.out.println("envelope:"+envelope.getExchange());
// System.out.println("envelope:"+envelope.getRoutingKey());
// System.out.println("properties:"+properties);
System.out.println("body:"+new String(body));
System.out.println("将日志打印控制台；");
}
};
channel.basicConsume(queue2Name,true,comsumer);
}
}
在配置routingkey的时候可以使用通配符，更加灵活

赞(0)
未经允许不得转载：171主机测评 » RabbitMQ的五大常用模式
分享到： 更多 (0)
标签：

相关推荐

评论 抢沙发

取消

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

热门标签

#人工智能(65154)#python(21700)#开发语言(18057)#java(16623)#运维(16356)#linux(15807)#大数据(15467)#服务器(13788)#算法(12415)#前端(11798)#ai(10494)#网络(10018)#数据库(9656)#c++(8779)#学习(7258)#课程设计(7058)#机器学习(7046)#区块链(6077)#后端(5621)#OpenClaw(5425)#spring boot(5361)#深度学习(5319)#c语言(4582)#安全(4580)#毕业设计(4415)#数据结构(4362)#架构(4237)#docker(4148)#javascript(4008)#大模型(3984)

热门文章

最新文章

归档

分类