【RabbitMQ】发布 / 订阅模式（使用案例）

文章目录

• 1. Publish / Subscribe（发布 / 订阅）
• 2、引入依赖
• 3、生产者代码编写
• • 3.1 创建交换机
  • 3.2 声明两个队列
  • 3.3 绑定队列和交换机
  • 3.4 发送消息
  • 3.5 完整代码
• 4、消费者代码编写
• • 4.1 消费者一
  • 4.1 消费者二
• 5. 观察结果

1. Publish / Subscribe（发布 / 订阅）

在发布 / 订阅模型中，多了一个 Exchange 角色。

Exchange 常见有三种类型，分别代表不同的路由规则：

• Fanout：广播，将消息交给所有绑定到交换机的队列（Publish / Subscribe 模式）
• Direct：定向，把消息交给符合指定 routing key 的队列（Routing 模式）
• Topic：通配符，把消息交给符合 routing pattern（路由模式）的队列（Topics 模式）

它们分别对应不同的工作模式，我们来看看 Publish / Subscribe 模式。

步骤：

• 1、引入依赖
• 2、编写生产者代码
• 3、编写消费者代码

2、引入依赖

先引入 rabbitmq 的依赖

<!– Source: https://mvnrepository.com/artifact/com.rabbitmq/amqp-client –>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.20.0</version>
<scope>compile</scope>
</dependency>

3、生产者代码编写

和前面两个的区别是：需要创建交换机，并且绑定队列和交换机。

3.1 创建交换机

需要用到下面的函数来创建

exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)

参数：

先在 Constants 函数中定义一个交换机和两个队列：

// 发布订阅模式
public static final String FANOUT_EXCHANGE = "fanout.exchange"; // 声明交换机
public static final String FANOUT_QUEUE1 = "fanout.queue1"; // 声明队列
public static final String FANOUT_QUEUE2 = "fanout.queue2"; // 声明队列

接着在生产者中编写代码

// 3. 声明交换机（使用内置的交换机即可）
channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);

3.2 声明两个队列

后面验证是否两个队列都能收到消息。

// 4. 声明队列
// 如果没有一个这样的队列，会自动创建，如果有，则不创建
channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);
channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);

3.3 绑定队列和交换机

需要用到的函数如下：

queueBind(String queue, String exchange, String routingKey)

参数如下：

代码如下所示：

// 5. 绑定队列和交换机
channel.queueBind(Constants.FANOUT_QUEUE1, Constants.FANOUT_EXCHANGE, "");
channel.queueBind(Constants.FANOUT_QUEUE2, Constants.FANOUT_EXCHANGE, "");

3.4 发送消息

需要用到下面的函数：

basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body)

参数说明：

• Exchange：交换机名称
• routingKey：如果交换机类型为 fanout，且 routingkey 设置为 ""，表示每个消费者都可以收到全部信息。

代码如下所示：

// 6. 发送消息
String msg = "Hello fanout… ";
channel.basicPublish(Constants.FANOUT_EXCHANGE, "",null, msg.getBytes());
System.out.println("消息发送成功");

3.5 完整代码

如下所示：

package fanout;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import 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 factory = new ConnectionFactory();
factory.setHost(Constants.HOST); // MQ所在的服务器地址
factory.setPort(Constants.PORT); // 端口号
factory.setUsername(Constants.USERNAME); // 账号
factory.setPassword(Constants.PASSWORD); // 密码
factory.setVirtualHost(Constants.VIRTUAL_HOST); // 虚拟主机
Connection connection = factory.newConnection();

// 2. 开启 channel 通道
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();
}
}

4、消费者代码编写

交换机和队列的绑定关系及声明已经在生产方写完，所以消费者不需要再写了。这里只需要去掉声明队列的代码就可以了。

1、创建Channel

2、接收消息，并处理

4.1 消费者一

代码如下所示：

package fanout;

import com.rabbitmq.client.*;
import 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 factory = new ConnectionFactory();
factory.setHost(Constants.HOST); // MQ所在的服务器地址
factory.setPort(Constants.PORT); // 端口号
factory.setUsername(Constants.USERNAME); // 账号
factory.setPassword(Constants.PASSWORD); // 密码
factory.setVirtualHost(Constants.VIRTUAL_HOST); // 虚拟主机
Connection connection = factory.newConnection();

// 2. 开启 channel 通道
Channel channel = connection.createChannel();

// 3. 声明队列
channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);

// 4. 接收消息并消费
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 从队列中收到消息后, 就会执行的方法
@Override
public 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);

// 5. 不需要释放资源
}
}

4.1 消费者二

代码如下所示：

package fanout;

import com.rabbitmq.client.*;
import 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 factory = new ConnectionFactory();
factory.setHost(Constants.HOST); // MQ所在的服务器地址
factory.setPort(Constants.PORT); // 端口号
factory.setUsername(Constants.USERNAME); // 账号
factory.setPassword(Constants.PASSWORD); // 密码
factory.setVirtualHost(Constants.VIRTUAL_HOST); // 虚拟主机
Connection connection = factory.newConnection();

// 2. 开启 channel 通道
Channel channel = connection.createChannel();

// 3. 声明队列
channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);

// 4. 接收消息并消费
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 从队列中收到消息后, 就会执行的方法
@Override
public 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);

// 5. 不需要释放资源
}
}

5. 观察结果

先运行生产者，可以看到两个队列分别有了一条消息。

然后在 Exchange 中找到咱们自己创建的交换机 fanout.exchange

可以看到 fanout.exchange 中多了队列绑定关系。

然后运行 Consumer 1 代码：

然后运行 Consumer 2 代码：

可以看到，生产者只发送了 1 次消息，而两个消费者都收到了该消息。