RabbitMQ简介
2020-04-21 16:03:09来源:博客园 阅读 ()
RabbitMQ简介
RabbitMQ简介
官网也讲述的清楚而且还有例子,只不过是英文的,很多人看到英文就不明白说什么了吧,即使有翻译成中文,总觉得哪里怪怪的,有些翻译并不流畅。我还是支持多看官网,官网:https://www.rabbitmq.com/ 。下面是自己做的一点下笔记,有参考其他文档。如有什么不对的地方,希望大家能够告诉我,通过留言板,像消息队列一样,你发送消息,我接收消息。
一、消息队列(MQ)
- 什么是消息队列:即MQ,Message Queue。是一种应用程序对应用对应用程序的通信方式。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来连接他们。消息队列是典型的:生产者、消费者模型。生产者不断向消息队列中生产消息,消费者不断从队列中获取消息。【因为消息的生产和消费都是异步的,而且只关心消息的发送和接收,没有业务逻辑的侵入,这样就实现了生产者和消费者的解耦】
- AMQP和JMS
- AMQP:即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准(高级消息队列协议)。【限定了数据传输的格式和方式,跨语言跨平台,和http协议类似】。是应用层协议的一个开放标准,为面向消息的中间件设计。
- JMS:Java MessageService,实际上指JMS API。JMS是SUN公司早期提出的消息标准,旨在为Java应用提供统一的消息操作,包括create、send、recieve等。JMS已经成为java Enterprise Edition的一部分。
- 两者间的区别和联系:
- JMS是定义了统一的接口,来对消息操作进行统一;AMQP是通过规定协议来统一数据交互的格式。
- JMS限定了必须使用Java语言;AMQP只是协议,不规定实现方式,因此是跨语言的。
- JMS规定了两种消息模型;而AMQP的消息模型更加丰富。
- 常见MQ产品:
- ActiveMQ:基于JMS, Apache旗下的
- RabbitMQ:基于AMQP协议,erlang(一种通用的面向并发的编程语言)语言开发,稳定性好
- RocketMQ:基于JMS,阿里巴巴产品,目前交由Apache基金会
- Kafka:分布式消息系统,高吞吐量
二、RabbitMQ:
- 下载安装:先安装erlang,在安装rabbitmq。(这里不详细说明,如果有不是很清楚的,可以去搜索,相信很多伙伴已经分享过很多这样的博客啦)
- 启动步骤【window系统下的cmd命令行】:cd \RabbitMQ Server\rabbitmq_server-3.7.15\sbin(进入到你安装的rabbitmq目录的sbin目录) --> rabbitmq-plugins enable rabbitmq_management --> rabbitmq-server【浏览器访问窗口:http://localhost:15672】
- 五种消息模型:RabbitMQ提供了6中消息模型,但是第6种RPC,并不是MQ。其中3/4/5这三种属于订阅模式,只不过进行路由的方式不同。
- 基本消息模型:RabbitMQ是一个消息的代理者(Message broker):它接收消息并且传递消息。你可以认为它是一个邮局,当你投递邮件到一个邮箱,你很肯定邮递员终究会将邮件递交给你的收件人。与此类似,RabbitMQ可以是一个邮箱、邮局、同时还是邮递员。不同之处在于:RabbitMQ不是传递纸质邮件,而是二进制的数据。
- 问题:那么RabbitMQ怎么知道消息被接收了呢?(转换思维,即如何避免消息的丢失?)【答:消费者的消息确认机制Acknowlege,当消费者获取消息后,会向RabbitMQ发送回执ACK,告知消息已经被接收。不过这种回执ACK分两种情况:① 自动ACK:消息一旦被接收,消费者自动发送ACK;② 手动ACK:消息接收后,不会发送ACK,需要手动调用。很好理解,望名知意嘛,嘿嘿嘿!!!】。
-
public class ConnectionUtil { /** * 建立与RabbitMQ的链接 */ public static Connection getConnection() throws IOException, TimeoutException { // 定义连接工厂 ConnectionFactory factory = new ConnectionFactory(); // 设置服务地址 factory.setHost("127.0.0.1"); // 端口 factory.setPort(5672); // 设置账号信息,用户名、密码、vhost factory.setVirtualHost("/demo"); factory.setUsername("guest"); factory.setPassword("guest"); // 通过工厂获取连接 Connection connection = factory.newConnection(); return connection; } }
public class Send { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] args) throws IOException, TimeoutException { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 从连接中创建通道,使用通道才能完成消息相关的操作 Channel channel = connection.createChannel(); // 声明(创建)队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 消息内容 String message = "Hello word!"; // 向指定的队列中发送消息 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); // 关闭通道和连接 channel.close(); connection.close(); } }
public class Recv { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] args) throws Exception { // 获取连接 Connection connection = ConnectionUtil.getConnection(); // 创建通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息时,会自动被调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [x] received: " + msg + "!"); } }; // 监听队列,第二个参数:是否自动进行消息确认 channel.basicConsume(QUEUE_NAME, true, consumer); } }
public class Recv2 { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] args) throws Exception { // 获取连接 Connection connection = ConnectionUtil.getConnection(); // 创建通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息时,会自动被调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [x] received: " + msg + "!"); // 手动进行ACK channel.basicAck(envelope.getDeliveryTag(), false); } }; // 监听队列,第二个参数false,手动进行ACK channel.basicConsume(QUEUE_NAME, false, consumer); } }
- work消息模型:也称为Task queue任务模型。当消息处理比较耗时的时候,可能产生消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。此时就可以使用该模型:让多个消费者绑定到一个队列,共同消费队列中的消息。队列中的消息一旦消费,就会消失,因此任务是不会被重复执行的。
- 问题:又会存在疑惑,任务是如何分配的,可以和显示生活中任务的分配联想起来呦?【答案:① 默认是任务平分,一次分配完全(即公平分配);② channel.basicQos(int num); 设置每个消费者同时只能处理num条数据(即能者多劳,耗时小的多处理些,你懂滴)】。
-
public class Send { private final static String QUEUE_NAME = "task_work_queue"; public static void main(String[] args) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 从连接中创建通道,使用通道才能完成消息相关的操作 final Channel channel = connection.createChannel(); // 声明(创建)队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 循环发布任务 for (int i=0; i<50; i++) { // 消息内容 String message = "task ... " + i; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); Thread.sleep(i * 2); } // 关闭通道和连接 channel.close(); connection.close(); } }
public class Recv { private final static String QUEUE_NAME = "task_work_queue"; public static void main(String[] args) throws Exception { // 获取连接 Connection connection = ConnectionUtil.getConnection(); // 创建通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 设置每个消费同时只能处理一条消息 channel.basicQos(1); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息时,会自动被调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [x] received: " + msg + "!"); try { // 模拟完成任务的耗时:1000ms Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } channel.basicAck(envelope.getDeliveryTag(), false); } }; // 监听队列,第二个参数:是否自动进行消息确认 channel.basicConsume(QUEUE_NAME, false, consumer); } }
/* * 对比上个消费者:耗时小,完成任务多些 */ public class Recv2 { private final static String QUEUE_NAME = "task_work_queue"; public static void main(String[] args) throws Exception { // 获取连接 Connection connection = ConnectionUtil.getConnection(); // 创建通道 final Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 设置每个消费同时只能处理一条消息 channel.basicQos(1); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息时,会自动被调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [x] received: " + msg + "!"); // 手动进行ACK channel.basicAck(envelope.getDeliveryTag(), false); } }; // 监听队列,第二个参数false,手动进行ACK channel.basicConsume(QUEUE_NAME, false, consumer); } }
- 订阅模型分类:
- 订阅模型 - Fanout:广播。一条消息,会被所有订阅的队列消费。
-
public class Send { private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] args) throws Exception { // 获取连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange,指定类型为fanout channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); // 消息内容 String message = "Hello everyone"; // 发布消息到Exchange channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [生产者] Sent '" + message + "'"); channel.close(); connection.close(); } }
public class Recv { private final static String QUEUE_NAME = "fanout_exchange_queue_1"; private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] args) throws IOException, TimeoutException { // 获取到链接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 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); System.out.println(" [消费者1] received: " + msg + "!"); } }; // 监听队列,自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }
public class Recv2 { private final static String QUEUE_NAME = "fanout_exchange_queue_2"; private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] args) throws IOException, TimeoutException { // 获取到链接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 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); System.out.println(" [消费者2] received: " + msg + "!"); } }; // 监听队列,自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }
-
- 订阅模型 - Direct:不同的消息被不同的队列消费。在Direct模型下:
- 队列与交换机的绑定,不能是任意绑定,而是要指定至少一个 RoutingKey(路由key);
- 消息的发送方向 Exchange 发送消息时,也必须指定消息的 RoutingKey;
- Exchange 不再把消息交给每一个绑定的队列,而是根据消息的 RoutingKey 进行判断,只有队列的RoutingKey与消息的RoutingKey完全一致,才会接收到消息。
-
public class Send { private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] args) throws Exception { // 获取连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange,指定类型为direct channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT); // 消息内容 String message = "商品增加了,id = 1002"; // 发布消息到Exchange,并且指定routing key为:delete,代表删除商品 channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes()); System.out.println(" [商品服务] Sent '" + message + "'"); channel.close(); connection.close(); } }
public class Recv { private final static String QUEUE_NAME = "direct_exchange_queue_1"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] args) throws IOException, TimeoutException { // 获取到链接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定义队列的消费者 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); System.out.println(" [消费者1] received: " + msg + "!"); } }; // 监听队列,自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }
public class Recv2 { private final static String QUEUE_NAME = "direct_exchange_queue_2"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] args) throws IOException, TimeoutException { // 获取到链接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert"); // 定义队列的消费者 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); System.out.println(" [消费者2] received: " + msg + "!"); } }; // 监听队列,自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }
- 订阅模型 - Topic:可以根据RoutingKey把消息路由到不同的队列,Topic类型Exchange可以让队列在绑定RoutingKey的时候使用通配符。
- RoutingKey 一般都是有一个或多个单词组成,多个单词之间以“ . ”分割,例:item.insert。
- 通配符规则:
- #:匹配一个或多个词
- *:匹配不多不少恰好一个词
-
public class Send { private final static String EXCHANGE_NAME = "topic_durable_exchange_test"; public static void main(String[] args) throws Exception { // 获取连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 开启生产者确认 // channel.confirmSelect(); // 声明exchange,指定类型为topic, 并且设置durable为true,持久化 channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, true); // 消息内容 String message = "商品新增了,id = 1002"; // 发布消息到Exchange,并且指定routing key,消息持久化 channel.basicPublish(EXCHANGE_NAME, "item.insert", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes()); System.out.println(" [商品服务] Sent '" + message + "'"); // 等待rabbitmq的确认消息,true为确认收到,false为发出有误 // channel.waitForConfirms(); channel.close(); connection.close(); } }
public class Recv { private final static String QUEUE_NAME = "topic_durable_exchange_queue_1"; private final static String EXCHANGE_NAME = "topic_durable_exchange_test"; public static void main(String[] args) throws IOException, TimeoutException { // 获取到链接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列, 第二个参数:true代表声明为持久化 channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete"); // 定义队列的消费者 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); System.out.println(" [消费者1] received: " + msg + "!"); } }; // 监听队列,自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }
public class Recv2 { private final static String QUEUE_NAME = "topic_durable_exchange_queue_2"; private final static String EXCHANGE_NAME = "topic_durable_exchange_test"; public static void main(String[] args) throws IOException, TimeoutException { // 获取到链接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*"); // 定义队列的消费者 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); System.out.println(" [消费者2] received: " + msg + "!"); } }; // 监听队列,自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }
- 订阅模型 - Fanout:广播。一条消息,会被所有订阅的队列消费。
- 基本消息模型:RabbitMQ是一个消息的代理者(Message broker):它接收消息并且传递消息。你可以认为它是一个邮局,当你投递邮件到一个邮箱,你很肯定邮递员终究会将邮件递交给你的收件人。与此类似,RabbitMQ可以是一个邮箱、邮局、同时还是邮递员。不同之处在于:RabbitMQ不是传递纸质邮件,而是二进制的数据。
- 当然啦,国内这么hot的rabbitMQ自然也是有集成到了springboot中滴,开心。
- maven坐标依赖:
<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.0.4.RELEASE</version> </parent> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> </dependencies>
- application.yml:
spring: rabbitmq: host: 127.0.0.1 username: guest password: guest virtual-host: /demo
-
@Component public class Listener { @RabbitListener(bindings = @QueueBinding( value = @Queue(value = "spring.test.queue", durable = "true"), exchange = @Exchange( value = "spring.test.exchange", type = ExchangeTypes.TOPIC), key = {"#.#"})) public void listen(String msg) { System.out.println("接收到的消息: " + msg); } }
- maven坐标依赖:
- 持久化:要将消息持久化,前提是:队列、Exchange都持久化。
- 交换机持久化:channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, true); // 参数三:设置durable为true。
- 队列持久化:channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 参数二:设置为true,表示设置队列持久化。
- 消息持久化:channel.basicPublish(EXCHANGE_NAME, "item.insert", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
- 解决消息丢失?
- ACK(消费者确认,由消费者向mq发送,防止消息丢失于消费者)
- 持久化(防止rabbitmq把消息丢失)
- 生产者确认机制publisher confirm(由mq向生产者发送,有些mq包含,有些不包含,比如:activeMQ不包含该机制,rabbitmq包含该机制)
- 发送消息前,将消息持久化到数据库,并记录消息状态(可靠消息服务)
- 思考问题(这个问题就留给你们思考啦?冲啊):如何保证消息发送的重复性,如何保证接口的幂等性(同一接口被重复执行,其结果一致)?【提示:加标识 消息的重发要谨慎】
原文链接:https://www.cnblogs.com/sunxn77/p/12740630.html
如有疑问请与原作者联系
标签:
版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有
上一篇:JAVA基础知识
- Java 简介 2020-06-09
- 黑菜菌的JAVA学习笔记 2020-06-09
- JAVA8 Stream流 2020-06-02
- Redis入门实战(1)-简介 2020-05-31
- 微服务项目持续集成部署流程简介 2020-05-27
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