本文是作者原创,版权归作者所有.若要转载,请注明出处.
本文RabbitMQ版本为rabbitmq-server-3.7.17,erlang为erlang-22.0.7.请各位去官网查看版本匹配和下载,也可以留言,我发安装包
AMQP协议
AMQP 一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。
AMQP是一个二进制协议，拥有一些现代化特点：多信道、协商式，异步，安全，扩平台，中立，高效。
RabbitMQ是AMQP协议的Erlang的实现。
| 概念 | 说明 |
|---|---|
| 连接Connection | 一个网络连接，比如TCP/IP套接字连接。 |
| 会话Session | 端点之间的命名对话。在一个会话上下文中，保证“恰好传递一次”。 |
| 信道Channel | 多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。 |
| 客户端Client | AMQP连接或者会话的发起者。AMQP是非对称的，客户端生产和消费消息，服务器存储和路由这些消息。 |
| 服务节点Broker | 消息中间件的服务节点；一般情况下可以将一个RabbitMQ Broker看作一台RabbitMQ 服务器。 |
| 端点 | AMQP对话的任意一方。一个AMQP连接包括两个端点（一个是客户端，一个是服务器）。 |
| 消费者Consumer | 一个从消息队列里请求消息的客户端程序。 |
| 生产者Producer | 一个向交换机发布消息的客户端应用程序。 |
简单队列
1.创建工程,添加依赖
<dependencies>
<!--RabbitMQ的依赖-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>
</dependencies>
2. 编写生产者
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
static final String QUEUE_NAME = "simple_queue";
static final String Host = "192.168.75.163";
static final Integer Port = 5672;
static final String VirtualHost = "/hello";
static final String Username = "test01";
static final String Password = "test01";
public static void main(String[] args) throws Exception{
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//主机地址：默认为localhost
connectionFactory.setHost(Host);
//连接端口
connectionFactory.setPort(Port);
//虚拟主机名称：默认为 /
connectionFactory.setVirtualHost(VirtualHost);
//连接用户名：默认为guest
connectionFactory.setUsername(Username);
//连接密码： 默认为guest
connectionFactory.setPassword(Password);
//创建连接
Connection connection = connectionFactory.newConnection();
//创建频道
Channel channel = connection.createChannel();
//声明（创建）队列
/*
参数一：队列名称
参数二：是否定义持久化队列
参数三：是否独占本次连接
参数四：是否在不使用的时候自动删除队列
参数五：队列其他参数
*/
channel.queueDeclare(QUEUE_NAME,true,false,false,null);
//要发送的消息
String message = "hello，rabbitmq！";
/**
* 参数1：交换机名称，不能为null 如果没有指定则使用默认Default Exchage
* 参数2：路由key,简单模式可以传递队列名称
* 参数3：消息其它属性
* 参数4：消息内容
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("已发送消息："+message);
//释放资源
channel.close();
connection.close();
}
}
我们debug看下消息的产生过程
在这里Connection还没创建
继续下一行
此时Connection创建了,Channel还没创建
继续下一行
此时Channel创建了,Queue还没创建,继续跟
此时Queue创建了,msg还没内容,继续跟
此时msg已经有内容了,然后释放资源,结束.
此时connection和channel都关闭了,只有queue还在,且有一条消息
3. 编写消费者
抽取创建connection的工具类
import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class ConnectionUtil { private static final String QUEUE_NAME = "simple_queue"; private static final String Host = "192.168.75.163"; private static final Integer Port = 5672; private static final String VirtualHost = "/hello"; private static final String Username = "test01"; private static final String Password = "test01"; public static Connection getConnection() throws Exception { //创建连接工厂 ConnectionFactory connectionFactory = new ConnectionFactory(); //主机地址;默认为 localhost connectionFactory.setHost(Host); //连接端口;默认为 5672 connectionFactory.setPort(Port); //虚拟主机名称;默认为 / connectionFactory.setVirtualHost(VirtualHost); //连接用户名；默认为guest connectionFactory.setUsername(Username); //连接密码；默认为guest connectionFactory.setPassword(Password); //创建连接 return connectionFactory.newConnection(); } }
编写消息的消费者
import com.itheima.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer { public static void main(String[] args) throws Exception { //获取连接 Connection connection = ConnectionUtil.getConnection(); //创建频道 Channel channel = connection.createChannel(); //创建队列：并设置消息处理 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null); //监听消息 DefaultConsumer consumer = new DefaultConsumer(channel) { @Override /* consumerTag ：消息者标签，在channel.basicConsume时候可以指定 envelope: 消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送） properties: 消息属性 body： 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("接收到的消息：" + new String(body, "UTF-8")); System.out.println(""); System.out.println("================================================================"); System.out.println(""); } }; /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.QUEUE_NAME, true, consumer); //不应该关闭资源，应该一直监听消息 //channel.close(); //connection.close(); } }
这里就不debug看了,直接看结果
可以看到,消费者是不应该关闭资源的,并且消息已经消费到了
4. 小结
上述的入门案例中中其实使用的是官网如下的简单模式：
在上图的模型中，有以下概念：
- P：生产者，也就是要发送消息的程序
- C：消费者：消息的接受者，会一直等待消息到来。
- queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
在此案例中RabbitMQ运转流程：
-
生产者发送消息
- 生产者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker；
- 声明队列并设置属性；如是否排它，是否持久化，是否自动删除；
- 将路由键（空字符串）与队列绑定起来；
- 发送消息至RabbitMQ Broker；
- 关闭信道；
- 关闭连接；
-
消费者接收消息
- 消费者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker
- 向Broker 请求消费相应队列中的消息，设置相应的回调函数；
- 等待Broker回应闭关投递响应队列中的消息，消费者接收消息；
- 确认（ack，自动确认）接收到的消息；
- RabbitMQ从队列中删除相应已经被确认的消息；
- 关闭信道；
- 关闭连接；