RabbitMQ简介
1.MQ为Message Queue,消息队列是应用程和应用程序之间的通信方法。
2.RabbitMQ是一个开源的,在AMQP基础上完整的,可复用的企业消息系统。
3.支持主流的操作系统,Linux、Windows、MacOX等。
4.多种开发语言支持,Java、Python、Ruby、.NET、PHP、C/C++、node.js等
官方网站 :http://www.rabbitmq.com/
一、队列之简单队列:
P:消息的生产者
C:消息的消费者
红色:队列
生产者将消息发送到队列,消费者从队列中获取消息。
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>3.4.1</version> </dependency>
public class ConnectionUtil { public static Connection getConnection() throws Exception { //定义连接工厂 ConnectionFactory factory = new ConnectionFactory(); //设置服务地址 factory.setHost("localhost"); //端口 factory.setPort(5672); //设置账号信息,用户名、密码、vhost factory.setVirtualHost("/mytest"); factory.setUsername("mytest"); factory.setPassword("mytest"); // 通过工程获取连接 Connection connection = factory.newConnection(); return connection; } }
生产者发送消息到队列
public class Send { private final static String QUEUE_NAME = "test_queue"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); // 从连接中创建通道 Channel channel = connection.createChannel(); // 声明(创建)队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 消息内容 String message = "Hello World!"; 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 = "test_queue"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列 channel.basicConsume(QUEUE_NAME, true, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); } } }
二、队列之Work模式的能者多劳:
一个生产者、2个消费者。
一个消息只能被一个消费者获取。
消费者1
public class Recv { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); //休眠 Thread.sleep(10); // 返回确认状态 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
消费者2
public class Recv2 { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成状态 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); // 休眠1秒 Thread.sleep(1000); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
生产者
向队列中发送100条消息。
public class Send { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); for (int i = 0; i < 100; i++) { // 消息内容 String message = "" + i; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); Thread.sleep(i * 10); } channel.close(); connection.close(); } }
能者多劳,实现公平调度。
实现公平调度的方式就是让每个消费者在同一时刻会分配一个任务。 通过channel.basicQos(1);可以设置
// 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1);
消息的确认模式
消费者从队列中获取消息,服务端如何知道消息已经被消费呢?
模式1:自动确认
只要消息从队列中获取,无论消费者获取到消息后是否成功消费,都认为是消息已经成功消费。
模式2:手动确认
消费者从队列中获取消息后,服务器会将该消息标记为不可用状态,等待消费者的反馈,如果消费者一直没有反馈,那么该消息将一直处于不可用状态。
为了确保消息一定被消费者处理,rabbitMQ提供了消息确认功能,就是在消费者处理完任务之后,就给服务器一个回馈,服务器就会将该消息删除,
如果消费者超时不回馈,那么服务器将就将该消息重新发送给其他消费者。
手动模式:
public class Recv { private final static String QUEUE_NAME = "test_queue_work"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 boolean autoAck = false;//true : 自动确认 false:手动确认 channel.basicConsume(QUEUE_NAME, autoAck, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); //休眠 Thread.sleep(10); // 返回确认状态 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
自动模式:
代码片段
// 监听队列,手动返回完成 boolean autoAck = true;//true : 自动确认 false:手动确认 channel.basicConsume(QUEUE_NAME, autoAck, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); }
三、队列之订阅模式:
解读:
1、1个生产者,多个消费者
2、每一个消费者都有自己的一个队列
3、生产者没有将消息直接发送到队列,而是发送到了交换机
4、每个队列都要绑定到交换机
5、生产者发送的消息,经过交换机,到达队列,实现,一个消息被多个消费者获取的目的
RabbitMQ 三种Exchange:
Fanout Exchange
Direct Exchange
Topic Exchange
消息的生产者:
public class Send { private final static String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明exchange channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); // 消息内容 String message = "Hello World!"; channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }
注意:消息发送到没有队列绑定的交换机时,消息将会丢失,因为,交换机没有存储消息的能力,消息只能存在在队列中。
消费者1:
public class Recv { private final static String QUEUE_NAME = "test_queue_work"; private final static String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
消费者2:
public class Recv2 { private final static String QUEUE_NAME = "test_queue_work2"; private final static String EXCHANGE_NAME = "test_exchange_fanout"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
测试结果:
同一个消息被多个消费者获取。
四、队列之路由模式:
1. 消息传递时需要一个“RouteKey”,可以简单的理解为要发送到的队列名字。任何发送到Direct Exchange的消息都会被转发到RouteKey中指定的Queue。
2.如果vhost中不存在RouteKey中指定的队列名,则该消息会被抛弃。
图示:
生产者:
public class Send { private final static String EXCHANGE_NAME = "test_exchange_direct"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明exchange channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 消息内容 String message = "Hello World!"; channel.basicPublish(EXCHANGE_NAME, "key2", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }
消费者1:
public class Recv { private final static String QUEUE_NAME = "test_queue_work"; private final static String EXCHANGE_NAME = "test_exchange_direct"; public static void main(String[] argv) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }消费者2:
public class Recv2 { private final static String QUEUE_NAME = "test_queue_work2"; private final static String EXCHANGE_NAME = "test_exchange_direct"; public static void main(String[] argv) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
五、队列之通配符模式:
Topic Exchange :将路由键和某模式进行匹配。此时队列需要绑定到一个模式上。符号“#”匹配一个或多个词,符号“*”匹配不多不少一个词。
因此“audit.#”能够匹配到“audit.irs.corporate”,但是“audit.*”只会匹配到“audit.irs”
图示:
任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上
1.这种模式较为复杂,简单来说,就是每个队列都有其关心的主题,所有的消息都带有一个“标题”(RouteKey),Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。
2.这种模式需要RouteKey,也许要提前绑定Exchange与Queue。
3.在进行绑定时,要提供一个该队列关心的主题,如“#.log.#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。
4.“#”表示0个或若干个关键字,“*”表示一个关键字。如“log.*”能与“log.warn”匹配,无法与“log.warn.timeout”匹配;但是“log.#”能与上述两者匹配。
5.同样,如果Exchange没有发现能够与RouteKey匹配的Queue,则会抛弃此消息。
生产者:
public class Send { private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明exchange channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 消息内容 String message = "Hello World!"; channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); } }
消费者1:
public class Recv { private final static String QUEUE_NAME = "test_queue_topic_work"; private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 //channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.*"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
消费者2:
public class Recv2 { private final static String QUEUE_NAME = "test_queue_topic_work2"; private final static String EXCHANGE_NAME = "test_exchange_topic"; public static void main(String[] argv) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 //channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.#"); // 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1); // 定义队列的消费者 QueueingConsumer consumer = new QueueingConsumer(channel); // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, false, consumer); // 获取消息 while (true) { QueueingConsumer.Delivery delivery = consumer.nextDelivery(); String message = new String(delivery.getBody()); System.out.println(" [x] Received '" + message + "'"); Thread.sleep(10); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } } }
相关推荐
这个“rabbitMQ代码案例 简单入门”的资料包为初学者提供了了解和学习RabbitMQ的基础知识。 首先,让我们了解一下RabbitMQ的基本概念: 1. **消息队列**:RabbitMQ的核心是消息队列,它负责存储和转发消息。生产者...
【RabbitMQ极速入门】 RabbitMQ是一款广泛使用的开源消息队列系统,它基于Advanced Message Queuing Protocol(AMQP)标准,提供可靠、灵活的消息传递服务。作为一个分布式系统的支柱,RabbitMQ允许不同组件之间...
【RabbitMQ 入门到精通】:RabbitMQ 是一款流行的消息中间件,它基于 AMQP(Advanced Message Queuing Protocol)协议实现,用于在分布式系统中高效地传输消息,从而实现异步处理、解耦和流量控制。本教程旨在帮助...
【标题】:“rabbitMQ入门” 在IT行业中,消息队列是一种常见的中间件技术,用于解耦应用程序组件,提高系统的可扩展性和可靠性。RabbitMQ是一个开源的消息代理和队列服务器,广泛应用于分布式系统中。本篇文章将带...
【RabbitMQ快速入门】 RabbitMQ是一款广泛使用的开源消息队列系统,它基于AMQP(Advanced Message Queuing Protocol)协议,提供高可用性、可靠性和可扩展性,是分布式系统中实现异步任务处理、解耦组件以及负载...
消息中间件之RabbitMQ入门讲解”的主题中,我们将深入理解RabbitMQ的核心概念,如何通过控制台进行管理,以及如何在Spring Cloud框架下创建消息生产者和消费者。 首先,让我们了解RabbitMQ的基本概念。RabbitMQ的...
### RabbitMQ基础知识与应用 #### 一、RabbitMQ简介 RabbitMQ 是一款开源的消息代理软件,也是 AMQP(Advanced Message Queuing Protocol)标准的一个实现。它支持多种消息传递模式,包括点对点(Direct)、发布/...
### RabbitMQ 入门教程(JAVA) #### 一、RabbitMQ 概述 RabbitMQ 是一个消息中间件,其主要功能是接收来自生产者的消息,并根据规则将其路由、缓冲以及持久化后传递给消费者。RabbitMQ 和消息传递系统通常会使用...
在这个“RabbitMQ入门代码”中,我们将深入探讨如何使用Java来与RabbitMQ进行交互,包括队列持久化、消息持久化、Direct交换机、Fanout交换机和Topic交换机的基础测试代码。 首先,让我们从基础开始,了解如何在...
【RabbitMQ入门操作手册】提供了全面的RabbitMQ学习指南,从基础概念到实际操作,帮助初学者快速掌握这个强大的消息队列系统。RabbitMQ是一个基于AMQP(Advanced Message Queuing Protocol)的开源消息代理,其核心...
最近整理学习的RabbitMQ入门Dome,文件是一个普通java项目导入完成后在lib文件夹中amqp-client-5.2.0.jar,slf4j-api-1.7.25.jar添加进去即可,里面有5个dome分是 dome1 : 简单队列,dome2 :work模式,dome3 : 订阅...
RabbitMQ是一款开源的消息中间件,它基于AMQP(Advanced Message Queuing Protocol)协议,用于在分布式系统中高效地路由和传递消息。本教程将带你走进RabbitMQ的世界,了解其基本概念、安装过程,以及如何创建消息...
### RabbitMQ Java客户端入门知识点详解 #### 一、RabbitMQ概述 - **定义与功能**:RabbitMQ是一款开源消息代理软件,基于Erlang语言开发,支持多种消息协议,如AMQP(Advanced Message Queuing Protocol)。其...
**RabbitMQ开发入门** RabbitMQ是一款开源的消息队列系统,基于AMQP(Advanced Message Queuing Protocol)协议实现,广泛应用于分布式系统中,用于解耦应用程序、处理异步任务和实现消息通信。在Java开发中,...
### RabbitMQ入门知识点详解 #### 一、RabbitMQ简介 RabbitMQ是一款开源的消息中间件,基于Erlang语言开发而成。它支持多种消息发布/订阅模式,并且能够跨多平台运行。作为消息中间件,RabbitMQ的核心功能是接受、...
在上述简单的示例中,使用了默认的 Direct 交换器,它将消息直接投递到指定的队列。 6. **队列与持久化** 队列是存储消息的容器,可以设置队列是否持久化,以保证即使在 RabbitMQ 重启后,消息也不会丢失。在示例...
RabbitMQ是一款开源的消息队列系统,基于AMQP(Advanced Message Queuing Protocol)协议实现,广泛应用于分布式系统中,用于处理异步任务、解耦组件以及实现消息传递。本示例将带你逐步了解如何在你的项目中使用...
通过这个“RabbitMQ入门视频”,初学者将能够建立起对RabbitMQ基本操作和用法的全面认识,为进一步深入学习和实际项目应用打下坚实的基础。在后续的学习中,还可以探索RabbitMQ的高级特性,如死信队列、延迟队列、...
在本项目"Dubbo+RabbitMQ+SpringMvc+Maven简单Demo"中,开发者整合了四个关键的技术组件,旨在提供一个快速开发的基础架构。以下是这些技术的详细解释和相关知识点: 1. **Dubbo**: Dubbo是阿里巴巴开源的一个高...
**RabbitMQ系列(一)入门Helloworld** RabbitMQ是一种开源的消息代理和队列服务器,它基于AMQP(Advanced Message Queuing Protocol)协议,广泛应用于分布式系统中的异步处理和解耦。在这个系列的第一部分,我们...