详解RabbitMQ工作模式之发布订阅模式

目录

发布订阅模式

概念

概念介绍

特点和优势

应用场景

注意事项

代码案例

引入依赖

常量类

编写生产者代码

编写消费者1代码

运行代码

---

发布订阅模式

概念

RabbitMQ的发布订阅模式（Publish/Subscribe）是一种消息传递模式，它允许消息生产者（Publisher）将消息发送到交换机（Exchange），然后交换机根据路由规则将消息广播到一个或多个队列，最后由消费者（Subscriber）从队列中接收并处理消息。

图中X表⽰交换机, 在订阅模型中，多了⼀个Exchange⻆⾊, 过程略有变化。

概念介绍

Exchange: 交换机 (X).
作⽤: ⽣产者将消息发送到Exchange, 由交换机将消息按⼀定规则路由到⼀个或多个队列中(上图中⽣产者将消息投递到队列中, 实际上这个在RabbitMQ中不会发⽣. )
RabbitMQ交换机有四种类型: fanout,direct, topic, headers, 不同类型有着不同的路由策略. AMQP协议⾥还有另外两种类型, System和⾃定义, 此处不再描述.

1. Fanout:⼴播，将消息交给所有绑定到交换机的队列(Publish/Subscribe模式)
2. Direct:定向，把消息交给符合指定routing key的队列(Routing模式)
3. Topic:通配符，把消息交给符合routing pattern(路由模式)的队列(Topics模式)
4. headers类型的交换器不依赖于路由键的匹配规则来路由消息, ⽽是根据发送的消息内容中的headers属性进⾏匹配. headers类型的交换器性能会很差,⽽且也不实⽤,基本上不会看到它的存在.

Exchange(交换机）只负责转发消息, 不具备存储消息的能⼒, 因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息就会丢失
RoutingKey: 路由键.⽣产者将消息发给交换器时, 指定的⼀个字符串, ⽤来告诉交换机应该如何处理这个消息.
Binding Key:绑定. RabbitMQ中通过Binding(绑定)将交换器与队列关联起来, 在绑定的时候⼀般会指定⼀个Binding Key, 这样RabbitMQ就知道如何正确地将消息路由到队列了.

⽐如下图: 如果在发送消息时, 设置了RoutingKey 为orange, 消息就会路由到Q1

当消息的Routing key与队列绑定的Bindingkey相匹配时，消息才会被路由到这个队列.

BindingKey其实也属于路由键中的⼀种, 官⽅解释为:the routingkey to use for the binding.
可以翻译为:在绑定的时候使⽤的路由键. ⼤多数时候,包括官⽅⽂档和RabbitMQJava API 中都把
BindingKey和RoutingKey看作RoutingKey, 为了避免混淆,可以这么理解:

1. 在使⽤绑定的时候,需要的路由键是BindingKey.
2. 在发送消息的时候,需要的路由键是RoutingKey

特点和优势

1.解耦合：生产者和消费者之间通过交换机进行解耦。生产者无需知道消息将被传递到哪些队列，消费者也无需知道消息来自哪个生产者。这种解耦合使得系统更加灵活和可扩展。
2.多播：支持多个消费者同时处理同一条消息，实现消息的多播效果。这有助于提高系统的并行处理能力和容错性。
3.灵活性：可以根据需要使用不同类型的交换机和绑定规则，以满足不同的消息传递需求。RabbitMQ提供了多种交换机类型，如直接交换机、扇形交换机、主题交换机等。

应用场景

发布订阅模式适用于需要将消息广播给多个消费者的场景，例如：

1.实时通知：如系统状态更新、订单状态变更等实时事件的通知。通过发布订阅模式，可以将这些事件广播给所有感兴趣的消费者。
2.日志记录：将应用程序的日志信息广播到多个日志处理服务进行处理和存储。这有助于实现日志的集中管理和分析。
3.事件处理：在事件驱动架构中，将事件作为消息发布到交换机，由多个消费者订阅并处理这些事件。这有助于实现事件的异步处理和分布式处理。

注意事项

1.消息持久化：为了确保消息在RabbitMQ服务器重启后不会丢失，可以将消息和队列标记为持久性。这样，即使服务器发生故障，消息仍然可以被消费者接收和处理。
2.消息确认：RabbitMQ支持消息确认机制，确保消息在成功处理后才会从队列中删除。这有助于防止消息丢失和重复处理。
3.负载均衡：在发布订阅模式中，多个消费者可以监听同一个队列或不同的队列。为了实现负载均衡，可以配置多个消费者来处理同一个队列中的消息。

代码案例

引入依赖

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.21.0</version>
</dependency>

常量类

public class Constants {public static final String HOST = "47.98.109.138";public static final int PORT = 5672;public static final String USER_NAME = "study";public static final String PASSWORD = "study";public static final String VIRTUAL_HOST = "aaa";//发布订阅模式public static final String FANOUT_EXCHANGE = "fanout.exchange";public static final String FANOUT_QUEUE1 = "fanout.queue1";public static final String FANOUT_QUEUE2 = "fanout.queue2";
}

编写生产者代码

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT); //需要提前开放端口号connectionFactory.setUsername(Constants.USER_NAME);//账号connectionFactory.setPassword(Constants.PASSWORD);  //密码connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明交换机channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);//4. 声明队列channel.queueDeclare(Constants.FANOUT_QUEUE1,true,false,false,null);channel.queueDeclare(Constants.FANOUT_QUEUE2,true,false,false,null);//5. 交换机和队列绑定channel.queueBind(Constants.FANOUT_QUEUE1,Constants.FANOUT_EXCHANGE,"");channel.queueBind(Constants.FANOUT_QUEUE2,Constants.FANOUT_EXCHANGE,"");//6. 发布消息String msg = "hello fanout....";channel.basicPublish(Constants.FANOUT_EXCHANGE,"", null, msg.getBytes());System.out.println("消息发送成功");//7. 释放资源channel.close();connection.close();}
}

编写消费者1代码

import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT); //需要提前开放端口号connectionFactory.setUsername(Constants.USER_NAME);//账号connectionFactory.setPassword(Constants.PASSWORD);  //密码connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明队列channel.queueDeclare(Constants.FANOUT_QUEUE1,true,false,false,null);//4. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息, 就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息:"+ new String(body));}};channel.basicConsume(Constants.FANOUT_QUEUE1, true, consumer);}
}

编写消费者2代码

import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT); //需要提前开放端口号connectionFactory.setUsername(Constants.USER_NAME);//账号connectionFactory.setPassword(Constants.PASSWORD);  //密码connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明队列channel.queueDeclare(Constants.FANOUT_QUEUE2,true,false,false,null);//4. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息, 就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息:"+ new String(body));}};channel.basicConsume(Constants.FANOUT_QUEUE2, true, consumer);}
}

运行代码

运行生产者代码之后，可以在管理界面看到两个队列中都有1条消息。

由此我们可以看到，交换机收到的生产者生产的一条消息被广播到了两个队列，消费者都能够分别从这两个队列中得到一条消息并消费。