概述:
在上一章节介绍的工作模式中,我们的消费会进行轮询发送给所有的消息消费者,每个消费者接受消息之和为全部消息。本章节介绍的订阅发布者模式则为:将消息传递给所有的消息消费者,每个消费者都能接受到全部的消息。并且在订阅发布章节我们将新引入一个新的概念,交换机(Exchange)概念。
发布者将运行结果通知订阅者,只需要引入rabbitMq的依赖即可
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>RabbitMQ</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version></dependency></dependencies></project>
java代码实现如下:
package com.xiaohui.rabbitmq.publishSubscribe;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.xiaohui.rabbitmq.utils.ConnectionUtils;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {//交换机名称public static final String FANOUT_EXCHANGE = "fanout_exchange";//队列1public static final String FANOUT_QUEUE_1 = "fanout_queue_1";//队列2public static final String FANOUT_QUEUE_2 = "fanout_queue_2";public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection = ConnectionUtils.getConnection();//创建信道Channel channel = connection.createChannel();//声明交换机/*** 参数一:交换机名称* 参数二:交换机类型:fanout \topic \direct*/channel.exchangeDeclare(Producer.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);/*** 声明队列* durabe:持久化的消息,mq重启后消息仍在。* exclusive :独占的,一个消息队列独占一个连接*/channel.queueDeclare(Producer.FANOUT_QUEUE_1,true,false,false,null);channel.queueDeclare(Producer.FANOUT_QUEUE_2,true,false,false,null);//队列绑定交换机channel.queueBind(Producer.FANOUT_QUEUE_1,Producer.FANOUT_EXCHANGE,"");channel.queueBind(Producer.FANOUT_QUEUE_2,Producer.FANOUT_EXCHANGE,"");for (int i = 1; i < 5; i++) {String msg = "订阅发布模式小兔子来啦........"+i;channel.basicPublish(Producer.FANOUT_EXCHANGE,"",null,msg.getBytes());}channel.close();connection.close();}
}
java代码实现如下:
package com.xiaohui.rabbitmq.publishSubscribe;import com.rabbitmq.client.*;
import com.xiaohui.rabbitmq.utils.ConnectionUtils;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Cunsumer1 {public static void main(String[] args) throws IOException, TimeoutException {//创建消费端链接Connection connection = ConnectionUtils.getConnection();//创建消费端渠道Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(Producer.FANOUT_EXCHANGE,BuiltinExchangeType.FANOUT);//声明消费队列channel.queueDeclare(Producer.FANOUT_QUEUE_1, true,false,false,null);//队列绑定到交换机channel.queueBind(Producer.FANOUT_QUEUE_1,Producer.FANOUT_EXCHANGE,"");//监听消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("==================订阅消费者1开始===================");System.out.println("路由的key为:"+envelope.getRoutingKey());System.out.println("交换机为:"+envelope.getExchange());System.out.println("消息ID为:"+envelope.getDeliveryTag());System.out.println("收到的消息为:"+new String(body,"UTF-8"));System.out.println("===================订阅消费者1结束==================");try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}};/*** 第二个参数表示是否 向mqserver自动回复收到* 第三个参数表示消息回调*/channel.basicConsume(Producer.FANOUT_QUEUE_1,true,consumer);}
}
订阅者2 代码实现:
package com.xiaohui.rabbitmq.publishSubscribe;import com.rabbitmq.client.*;
import com.xiaohui.rabbitmq.utils.ConnectionUtils;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Cunsumer2 {public static void main(String[] args) throws IOException, TimeoutException {//创建消费端链接Connection connection = ConnectionUtils.getConnection();//创建消费端渠道Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(Producer.FANOUT_EXCHANGE,BuiltinExchangeType.FANOUT);//声明消费队列channel.queueDeclare(Producer.FANOUT_QUEUE_2, true,false,false,null);//队列绑定到交换机channel.queueBind(Producer.FANOUT_QUEUE_2,Producer.FANOUT_EXCHANGE,"");//监听消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("==================订阅消费者2开始===================");System.out.println("路由的key为:"+envelope.getRoutingKey());System.out.println("交换机为:"+envelope.getExchange());System.out.println("消息ID为:"+envelope.getDeliveryTag());System.out.println("收到的消息为:"+new String(body,"UTF-8"));System.out.println("===================订阅消费者2结束==================");try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}};/*** 第二个参数表示是否 向mqserver自动回复收到* 第三个参数表示消息回调*/channel.basicConsume(Producer.FANOUT_QUEUE_2,true,consumer);}
}
rabbitmq教程。运行打印(两个控制台一样):
==================订阅消费者1开始===================
路由的key为:
交换机为:fanout_exchange
消息ID为:10
收到的消息为:订阅发布模式小兔子来啦........1
===================订阅消费者1结束==================
==================订阅消费者1开始===================
路由的key为:
交换机为:fanout_exchange
消息ID为:11
收到的消息为:订阅发布模式小兔子来啦........2
===================订阅消费者1结束==================
==================订阅消费者1开始===================
路由的key为:
交换机为:fanout_exchange
消息ID为:12
收到的消息为:订阅发布模式小兔子来啦........3
===================订阅消费者1结束==================
==================订阅消费者1开始===================
路由的key为:
交换机为:fanout_exchange
消息ID为:13
收到的消息为:订阅发布模式小兔子来啦........4
===================订阅消费者1结束==================
版权声明:本站所有资料均为网友推荐收集整理而来,仅供学习和研究交流使用。
同时也捎带提醒一句,如果你发现你 hook 系统的方法不起作用的时候,或许可以检查一下你项目里引入的第三方框架里是否也 hook 了和你一样的" alt="从使用 KVO 监听 readonly 属性说起">
工作时间:8:00-18:00
客服电话
电子邮件
admin@qq.com
扫码二维码
获取最新动态