第四章-RabbitMQ工作模式-PUB/SUB
1.模式介绍
1.1 模式
此模式称为发布订阅模式，从此模式开始，我们就不再使用默认的交换机了，开始定义我们自己的交换机。
此发布订阅模式，使用的交换机类型为Fanout。定义好交换机，消息的传输路径变为，从Producer发出，发送至Fanout类型的交换机，交换机将消息分别推送给和自己绑定的Queue上，
消费者再去消费对应的Queue，此处与前面模式一致。
1.2 应用场景
此种模式用在什么场景呢，凡是场景中触发一个动作后，后面的流程都需要根据此动作做出反应，而后面的流程一般>1个，流程之间也没有先后顺序，还需要快速执行，即可以使用此模式。
举个栗子：
1. 用户注册成功后发送通知，既要发短信，又发邮件，还发站内信，发信息这些流程也不存在先后顺序，即可以使用这种模式。用户注册后，推送消息到mq， 发短信、发邮件、发站内信分别去消费消息实现信息发送。
2. 支付系统中用户支付成功后，后续要有一系列动作，例如要上B端账，调风控、结算、发消息等逻辑，也同样可以使用此模式应用其中。
1.2 模拟
首先注意到我们的Exchange 类型 是 fanout，
其次在发送的时候 routingkey -填与不填都不再影响 Exchange的分发路由了。
2.验证代码
还是举例 小明洗澡的例子。小明一天在看科幻小说，突然睡着了，梦里梦见自己研究出了一种热水器转换开关，这种开关可以将流过的水复制一份。心想太棒了，这以后能省不少水。
于是改造了下自己的太阳能热水器，热水器内部变成了两个水槽，加入此转换器后，分别给两个水槽上水，水管出水1L，两个水槽都能充满1L。改造成功后，便开始拉着自己的女朋友开始了洗澡实验。
2.1 生产者
/**
* @author rabbit
* @version 1.0.0
* @Description 发布订阅模式一个生产者，一个交换机，两个队列，两个消费者
* 声明一个Fanout类型的exchange，并且将exchange和queue绑定在一起，绑定规则直接绑定。
* 生产者创建一个exchange并且指定类型，和一个或多个队列绑定在一起。当生产者发送消息是会发送到exchange中，再由exchange到绑定的队列中
* @createTime 2022/07/27 19:34:00
*/
public class WaterProducer {
public static final String PUBSUB_QUEUE_1 = "SolarWaterHeaterXM";
public static final String PUBSUB_QUEUE_2 = "SolarWaterHeaterXL";
//生产者
public static void main(String[] args) throws Exception {
//1、获取connection
Connection connection = RabbitCommonConfig.getConnection();
//2、创建channel
Channel channel = connection.createChannel();
for (int i = 1; i <= 10; i++) {
sendMsg(channel, i);
Thread.sleep(100);
}
//4、关闭管道和连接
channel.close();
connection.close();
}
private static void sendMsg(Channel channel, int k) throws IOException {
//3、通过channel创建自己的exchange 并且绑定队列
/**
* 参数1：exchange的名称
* 参数2：指定exchange的类型
* FANOUT-Publish/Subscribe
* DIRECT-Routing
* TOPIC-Topics
*/
channel.exchangeDeclare("publish-exchange", BuiltinExchangeType.FANOUT);
channel.queueBind(PUBSUB_QUEUE_1, "publish-exchange", "");
channel.queueBind(PUBSUB_QUEUE_2, "publish-exchange", "");
//4、发送消息到exchange
String msg = k + "升";
/**
* 参数1：指定exchange，使用“”。默认的exchange
* 参数2：指定路由的规则，使用具体的队列名称。exchange为""时，消息直接发送到队列中
* 参数3：指定传递的消息携带的properties
* 参数4：指定传递的消息，byte[]类型
*/
channel.basicPublish("publish-exchange", "", null, msg.getBytes());
System.out.println("水龙头放水成功！" + k + "升");
}
}2.2 消费者
小明：
/**
* @author rabbit
* @version 1.0.0
* @Description 发布订阅模式 一个生产者，一个交换机，两个队列，两个消费者
* @createTime 2022/07/27 19:36:00
*/
public class XMShowerConsumer {
public static final String PUBSUB_QUEUE_1 = "SolarWaterHeaterXM";
//消费者
public static void main(String[] args) throws Exception {
//1、获取连对象、
Connection connection = RabbitCommonConfig.getConnection();
//2、创建channel
Channel channel = connection.createChannel();
channel.basicQos(1);
//3、创建队列
/**
* 参数1：queue 指定队列名称
* 参数2：durable 是否开启持久化（true）
* 参数3：exclusive 是否排外（conn.close()-》当前对列自动删除，当前队列只能被一个 消费者消费）
* 参数4：autoDelete 如果这个队列没有其他消费者在消费，队列自动删除
* 参数5：arguments 指定队列携带的信息
*/
channel.queueDeclare(PUBSUB_QUEUE_1, true, false, false, null);
//4.开启监听Queue
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, "UTF-8"));
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//手动ACK（接收信息，指定是否批量操作）
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
/**
* 参数1：queue 指定消费哪个队列
* 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）
* 参数1：cancelCallback 指定消费回调
*
*/
//3.关闭自动ACK
channel.basicConsume(PUBSUB_QUEUE_1,false,consumer);
System.out.println("小明使用热水器中的XM水槽开始洗澡......");
//5、键盘录入，让程序不结束！
System.in.read();
//6、释放资源
channel.close();
connection.close();
}
}
小丽：
/**
* @author rabbit
* @version 1.0.0
* @Description 发布订阅模式 一个生产者，一个交换机，两个队列，两个消费者
* @createTime 2022/07/27 19:36:00
*/
public class XLShowerConsumer {
public static final String PUBSUB_QUEUE_2 = "SolarWaterHeaterXL";
//消费者
public static void main(String[] args) throws Exception {
//1、获取连对象、
Connection connection = RabbitCommonConfig.getConnection();
//2、创建channel
Channel channel = connection.createChannel();
channel.basicQos(1);
//3、创建队列-helloworld
/**
* 参数1：queue 指定队列名称
* 参数2：durable 是否开启持久化（true）
* 参数3：exclusive 是否排外（conn.close()-》当前对列自动删除，当前队列只能被一个 消费者消费）
* 参数4：autoDelete 如果这个队列没有其他消费者在消费，队列自动删除
* 参数5：arguments 指定队列携带的信息
*/
channel.queueDeclare(PUBSUB_QUEUE_2, true, false, false, null);
//4.开启监听Queue
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, "UTF-8"));
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//手动ACK（接收信息，指定是否批量操作）
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
/**
* 参数1：queue 指定消费哪个队列
* 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）
* 参数1：cancelCallback 指定消费回调
*
*/
//3.关闭自动ACK
channel.basicConsume(PUBSUB_QUEUE_2, false, consumer);
System.out.println("小丽使用热水器中的XL水槽开始洗澡......");
//5、键盘录入，让程序不结束！
System.in.read();
//6、释放资源
channel.close();
connection.close();
}
}2.3 结果
生产者：
水龙头放水成功！1升
水龙头放水成功！2升
水龙头放水成功！3升
水龙头放水成功！4升
水龙头放水成功！5升
水龙头放水成功！6升
水龙头放水成功！7升
水龙头放水成功！8升
水龙头放水成功！9升
水龙头放水成功！10升消费者小明：
小明洗澡用水: 1升
小明洗澡用水: 2升
小明洗澡用水: 3升
小明洗澡用水: 4升
小明洗澡用水: 5升
小明洗澡用水: 6升
小明洗澡用水: 7升
小明洗澡用水: 8升
小明洗澡用水: 9升
小明洗澡用水: 10升
消费者小丽：
小丽洗澡用水: 1升
小丽洗澡用水: 2升
小丽洗澡用水: 3升
小丽洗澡用水: 4升
小丽洗澡用水: 5升
小丽洗澡用水: 6升
小丽洗澡用水: 7升
小丽洗澡用水: 8升
小丽洗澡用水: 9升
小丽洗澡用水: 10升
最终实现了，生产者出水10L，两个人都使用了10L水洗了澡，刚洗完，小明就睡醒了，原来是一个梦。
3. 总结
从此模式开始，我们接触到了Exchange的创建，及绑定，以及使用了Exchange的类型 Fanout。
而以前我们在simple模式及work模式中，用到的默认Exchange类型都是Direct类型。
核心代码-声明Exchange：
channel.exchangeDeclare("publish-exchange", BuiltinExchangeType.FANOUT);核心代码-绑定queue
channel.queueBind(PUBSUB_QUEUE_1, "publish-exchange", "");核心代码-发送给Fanout的Exchange
channel.basicPublish("publish-exchange", "", null, msg.getBytes());
![[附源码]java毕业设计企业员工管理系统](https://img-blog.csdnimg.cn/d45e248882164a598c7456acaf91fc5e.png)
