RabbitMQ学习第三记:发布/订阅模式(Publish/Su…
2018-09-10 01:03:41来源:博客园 阅读 ()
工作队列模式是直接在生产者与消费者里声明好一个队列,这种情况下消息只会对应同类型的消费者。
举个用户注册的列子:用户在注册完后一般都会发送消息通知用户注册成功(失败)。如果在一个系统中,用户注册信息有邮箱、手机号,那么在注册完后会向邮箱和手机号都发送注册完成信息。利用MQ实现业务异步处理,如果是用工作队列的话,就会声明一个注册信息队列。注册完成之后生产者会向队列提交一条注册数据,消费者取出数据同时向邮箱以及手机号发送两条消息。但是实际上邮箱和手机号信息发送实际上是不同的业务逻辑,不应该放在一块处理。这个时候就可以利用发布/订阅模式将消息发送到转换机(EXCHANGE),声明两个不同的队列(邮箱、手机),并绑定到交换机。这样生产者只需要发布一次消息,两个队列都会接收到消息发给对应的消费者。
1、什么是发布/订阅模式(Publish/Subscribe)
简单解释就是,可以将消息发送给不同类型的消费者。做到发布一次,消费多个。下图取自于官方网站(RabbitMQ)的发布/订阅模式的图例
P:消息的生产者
X:交换机
红色:队列
C1,C2:消息消费者
下面是利用用户注册解释的该模式。(先运行两个消费者,在运行生产者。如果没有提前将队列绑定到交换机,那么直接运行生产者的话,消息是不会发到任何队列里的)
2、生产者(Send)代码
public class Send { private final static String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] args) { try { //获取连接 Connection connection = ConnectionUtil.getConnection(); //从连接中获取一个通道 Channel channel = connection.createChannel(); //声明交换机(分发:发布/订阅模式) channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //发送消息 for (int i = 0; i < 10; i++) { String message = "this is user registe message" + i; System.out.println("[send]:" + message); //发送消息 channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("utf-8")); Thread.sleep(5 * i); } channel.close(); connection.close(); } catch (Exception e) { e.printStackTrace(); } } }
运行结果:
[send]:this is user registe message0
[send]:this is user registe message1
[send]:this is user registe message2
[send]:this is user registe message3
[send]:this is user registe message4
[send]:this is user registe message5
[send]:this is user registe message6
[send]:this is user registe message7
[send]:this is user registe message8
[send]:this is user registe message9
3、消费者1(ReceiveEmail)
public class ReceiveEmail { //交换机名称 private final static String EXCHANGE_NAME = "test_exchange_fanout"; //队列名称 private static final String QUEUE_NAME = "test_queue_email"; public static void main(String[] args) { try { //获取连接 Connection connection = ConnectionUtil.getConnection(); //从连接中获取一个通道 final Channel channel = connection.createChannel(); //声明交换机(分发:发布/订阅模式) channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //将队列绑定到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); //保证一次只分发一个 int prefetchCount = 1; channel.basicQos(prefetchCount); //定义消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { //当消息到达时执行回调方法 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "utf-8"); System.out.println("[email] Receive message:" + message); try { //消费者休息2s处理业务 Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println("[1] done"); //手动应答 channel.basicAck(envelope.getDeliveryTag(), false); } } }; //设置手动应答 boolean autoAck = false; //监听队列 channel.basicConsume(QUEUE_NAME, autoAck, consumer); } catch (IOException e) { e.printStackTrace(); } } }
运行结果:
[email] Receive message:this is user registe message0
[1] done
[email] Receive message:this is user registe message1
[1] done
[email] Receive message:this is user registe message2
[1] done
[email] Receive message:this is user registe message3
[1] done
[email] Receive message:this is user registe message4
[1] done
[email] Receive message:this is user registe message5
[1] done
[email] Receive message:this is user registe message6
[1] done
[email] Receive message:this is user registe message7
[1] done
[email] Receive message:this is user registe message8
[1] done
[email] Receive message:this is user registe message9
[1] done
4、消费者2(ReceivePhone)
public class ReceivePhone { //交换机名称 private final static String EXCHANGE_NAME = "test_exchange_fanout"; //队列名称 private static final String QUEUE_NAME = "test_queue_phone"; public static void main(String[] args) { try { //获取连接 Connection connection = ConnectionUtil.getConnection(); //从连接中获取一个通道 final Channel channel = connection.createChannel(); //声明交换机(分发:发布/订阅模式) channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //将队列绑定到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); //保证一次只分发一个 int prefetchCount = 1; channel.basicQos(prefetchCount); //定义消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { //当消息到达时执行回调方法 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { String message = new String(body, "utf-8"); System.out.println("[phone] Receive message:" + message); try { //消费者休息2s处理业务 Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } finally { System.out.println("[2] done"); //手动应答 channel.basicAck(envelope.getDeliveryTag(), false); } } }; //设置手动应答 boolean autoAck = false; //监听队列 channel.basicConsume(QUEUE_NAME, autoAck, consumer); } catch (IOException e) { e.printStackTrace(); } } }
运行结果:
[phone] Receive message:this is user registe message0
[2] done
[phone] Receive message:this is user registe message1
[2] done
[phone] Receive message:this is user registe message2
[2] done
[phone] Receive message:this is user registe message3
[2] done
[phone] Receive message:this is user registe message4
[2] done
[phone] Receive message:this is user registe message5
[2] done
[phone] Receive message:this is user registe message6
[2] done
[phone] Receive message:this is user registe message7
[2] done
[phone] Receive message:this is user registe message8
[2] done
[phone] Receive message:this is user registe message9
[2] done
总结:
1、该模式下生产者并不是直接操作队列,而是将数据发送给交换机,由交换机将数据发送给与之绑定的队列。从运行结果中可以看到,两中类型的消费者(Email,Phone)都收到相同数量的消息。
2、该模式必须声明交换机,并且设置模式:channel.exchangeDeclare(EXCHANGE_NAME, "fanout") fanout指分发模式(将每一条消息都发送到与交换机绑定的队列。
3、 队列必须绑定交换机:channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
注意:本文仅代表个人理解和看法哟!和本人所在公司和团体无任何关系!
标签:
版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有
上一篇:关于面向对象
下一篇:设计模式-模板方法模式
- 学习Java 8 Stream Api (4) - Stream 终端操作之 collect 2020-06-11
- java学习之第一天 2020-06-11
- Java学习之第二天 2020-06-11
- Spring WebFlux 学习笔记 - (一) 前传:学习Java 8 Stream Ap 2020-06-11
- Linux简单命令的学习 2020-06-10
IDC资讯: 主机资讯 注册资讯 托管资讯 vps资讯 网站建设
网站运营: 建站经验 策划盈利 搜索优化 网站推广 免费资源
网络编程: Asp.Net编程 Asp编程 Php编程 Xml编程 Access Mssql Mysql 其它
服务器技术: Web服务器 Ftp服务器 Mail服务器 Dns服务器 安全防护
软件技巧: 其它软件 Word Excel Powerpoint Ghost Vista QQ空间 QQ FlashGet 迅雷
网页制作: FrontPages Dreamweaver Javascript css photoshop fireworks Flash