上文的工作队列模式是直接在生产者与消费者里声明好一个队列（不指定交换机，生产者和消费者之间直接连接队列），这种情况下消息只会对应同类型的消费者。
显然这种只处理同种类型的消息是有弊端的。
举个用户注册的列子
门户网站，用户在注册完后一般都会发送消息通知用户注册成功（失败）。
如果在一个系统中，用户注册信息有邮箱、手机号，那么在注册完后会向邮箱和手机号都发送注册完成信息（假设都发送）。
利用 MQ 实现业务异步处理，如果是用工作队列的话，就会声明一个注册信息队列。注册完成之后生产者会向队列提交一条注册数据，消费者取出数据同时向邮箱以及手机号发送两条消息。但是实际上邮箱和手机号信息发送实际上是不同的业务逻辑，不应该放在一块处理。(应该放到不同的队列中，由不同的消费者消费)
这个时候就可以利用发布/订阅模式将消息发送到转换机（EXCHANGE），声明两个不同的队列（邮箱、手机），并绑定到交换机。这样生产者只需要发布一次消息，两个队列都会接收到消息发给对应的消费者，大致如下图所示。
1. 什么是发布订阅模式
简单解释就是，可以将消息发送给不同类型的消费者。做到发布一次，消费多个。下图取自于官方网站（RabbitMQ）的发布/订阅模式的图例：
P 表示为生产者、 X 表示交换机、C1C2 表示为消费者，红色表示队列。
2.代码部分
生产者
cn.lovingliu.rabbitmq_fanout.producer.Producer
package cn.lovingliu.rabbitmq_fanout.producer;
import cn.lovingliu.common.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Author：LovingLiu
* @Description: 生产者
* @Date：Created in 2020-01-16
*/
public class Producer {
private static final String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
/** 1.创建新的连接 */
Connection connection = ConnectionUtil.getConnection();
/** 2.创建通道 */
Channel channel = connection.createChannel();
/** 3.绑定的交换机 参数1交互机名称 参数2 exchange类型 */
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
/** 4.发送消息 */
for (int i = 0; i < 10; i++)
{
String message = "用户注册消息：" + i;
System.out.println("[send]：" + message);
//发送消息
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("utf-8"));
try {
Thread.sleep(5 * i);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/** 5.关闭通道、连接 */
channel.close();
connection.close();
/** 注意：如果消费没有绑定交换机和队列，则消息会丢失 */
}
}
代码补充，channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("utf-8")); 其中第二个参数为空类似于表示全局广播，只要绑定到该队列上的消费者理论上是都可以收到的。
消费者
短信消费者cn.lovingliu.rabbitmq_fanout.consumer.ConsumerSMS
package cn.lovingliu.rabbitmq_fanout.consumer;
import cn.lovingliu.common.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Author：LovingLiu
* @Description: 短信消费者
* @Date：Created in 2020-01-16
*/
public class ConsumerSMS {
private static final String QUEUE_NAME = "ConsumerFanout_sms";
private static final String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
System.out.println("短信消费者启动");
/** 1.创建新的连接 */
Connection connection = ConnectionUtil.getConnection();
/** 2.创建通道 */
Channel channel = connection.createChannel();
/** 3.消费者关联队列 */
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/** 4.消费者绑定交换机 参数1 队列 参数2交换机 参数3 routingKey */
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String msg = new String(body, "UTF-8");
System.out.println("消费者获取生产者消息:" + msg);
}
};
/** 5.消费者监听队列消息 */
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
邮件消费者cn.lovingliu.rabbitmq_fanout.consumer.ConsumerEmail
package cn.lovingliu.rabbitmq_fanout.consumer;
import cn.lovingliu.common.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Author：LovingLiu
* @Description: 邮件消费
* @Date：Created in 2020-01-16
*/
public class ConsumerEmail {
private static final String QUEUE_NAME = "consumerFanout_email";
private static final String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
System.out.println("邮件消费者启动");
/** 1.创建新的连接 */
Connection connection = ConnectionUtil.getConnection();
/** 2.创建通道 */
Channel channel = connection.createChannel();
/** 3.消费者关联队列 */
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/** 4.消费者绑定交换机 参数1 队列 参数2交换机 参数3 routingKey */
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String msg = new String(body, "UTF-8");
System.out.println("消费者获取生产者消息:" + msg);
}
};
/** 5.消费者监听队列消息 */
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
代码补充， channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); 中第三个参数置为空时，可以接收到生产者所有的消息（生产者 routingKey参数为空时）。
3.运行截图
先运行两个消费者，再运行生产者。如果没有提前将队列绑定到交换机，那么直接运行生产者的话，消息是不会发到任何队列里的。
生产者
消费者
邮件消费者
短信消费者
总结
首先相对于工作模式，发布订阅模式引入了交换机的概念，相对其类型上更加灵活广泛一些。通过上文我们可以总结如下：
-
1.生产者不是直接操作队列，而是将数据发送给交换机，由交换机将数据发送给与之绑定的队列。从不加特定参数的运行结果中可以看到，两种类型的消费者（
email，sms）都收到相同数量的消息。 -
2.必须声明交换机，并且设置模式：
channel.exchangeDeclare(EXCHANGE_NAME, "fanout")，其中fanout指分发模式（将每一条消息都发送到与交换机绑定的队列）。 -
3.队列必须绑定交换机：
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
生产者发送消息到交换机，多个消费者声明多个队列，与交换机进行绑定，队列中的消息可以被所有消费者消费，类似于QQ群消息
网友评论