Rabbit快速入门
入门案例
需求：使用简单模式完成消息传递
步骤：
创建工程（生成者、消费者）
分别添加依赖
编写生产者发送消息
编写消费者接收消息
3.1.2. 添加依赖
往heima-rabbitmq的pom.xml文件中添加如下依赖：
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version></dependency>
3.2. 编写生产者
package com.itheima.producer;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 发送消息*/
public class Producer_HelloWorld {public static void main(String[] args) throws IOException, TimeoutException {///建立连接工厂ConnectionFactory factory = new ConnectionFactory();// 设置参数factory.setHost("localhost"); //默认值为localhostfactory.setPort(5672); //端口 默认值 5672factory.setVirtualHost("/chengpeng");// 虚拟机 默认 /factory.setUsername("chengpeng"); // 用户名 默认值 guestfactory.setPassword("chengpeng"); // 密码 默认值 guest/// 创建连接 connectionConnection connection = factory.newConnection();/// 创建ChannelChannel channel = connection.createChannel();///创建队列 声明队列/*** String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments* queue:队列名称* durable:是否持久化，当mq重启之后还在* exclusive：* 一：是否独占，只能有一个消费者监听队列* 二：当Connection关闭时，是否删除队列* autoDelete：是否自动删除 当没有Consumer时，自动删除* arguments：参数*//// 如果没有一个名字叫helloworld的队列，则会创建channel.queueDeclare("hello_world",true,false,false,null);//发送消息/*** String exchange, String routingKey, AMQP.BasicProperties props, byte[] body** exchange:交换机名称，简单模式下交换机会使用默认的""* routingKey:路由名称 如果给默认的交换机需要给队列名称一样* props:配置信息* body: 发送消息数据*/String body = "hello rabbitmq~~~~";channel.basicPublish("","hello_world",null,body.getBytes());/// 释放资源channel.close();connection.close();}
}
在执行上述的消息发送之后；可以登录rabbitMQ的管理控制台，可以发现队列和其消息：
3.3. 编写消费者
抽取创建connection的工具类com.itheima.rabbitmq.util.ConnectionUtil；
package com.itheima.consumer;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer_HelloWorld {public static void main(String[] args) throws IOException, TimeoutException {///建立连接工厂ConnectionFactory factory = new ConnectionFactory();// 设置参数factory.setHost("localhost"); //默认值为localhostfactory.setPort(5672); //端口 默认值 5672factory.setVirtualHost("/chengpeng");// 虚拟机 默认 /factory.setUsername("chengpeng"); // 用户名 默认值 guestfactory.setPassword("chengpeng"); // 密码 默认值 guest/// 创建连接 connectionConnection connection = factory.newConnection();/// 创建ChannelChannel channel = connection.createChannel();///创建队列 声明队列/*** String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments* queue:队列名称* durable:是否持久化，当mq重启之后还在* exclusive：* 一：是否独占，只能有一个消费者监听队列* 二：当Connection关闭时，是否删除队列* autoDelete：是否自动删除 当没有Consumer时，自动删除* arguments：参数*//// 如果没有一个名字叫helloworld的队列，则会创建channel.queueDeclare("hello_world",true,false,false,null);//接收消息/*** String queue, boolean autoAck, Consumer callback* queue: 队列名称* autoAck: 是否自动确认* callback: 回调对象*/Consumer consumer = new DefaultConsumer(channel){/*** 回调方法，当收到消息后，会自动执行该方法* @param consumerTag 标识* @param envelope 获取一些信息，交换机，路由key..* @param properties 配置信息* @param body 数据* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("consumerTag:"+consumerTag);System.out.println("Exchange:"+envelope.getExchange());System.out.println("RoutingKey:"+envelope.getRoutingKey());System.out.println("properties:"+properties);System.out.println("body:"+new String(body));}};channel.basicConsume("hello_world",true,consumer);/// 释放资源//channel.close();//connection.close();}
}
3.4. 小结
在上图的模型中，有以下概念：
- P：生产者，也就是要发送消息的程序
- C：消费者：消息的接受者，会一直等待消息到来。
- queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
4. AMQP
4.1. 相关概念介绍
AMQP 一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。
AMQP是一个二进制协议，拥有一些现代化特点：多信道、协商式，异步，安全，扩平台，中立，高效。
RabbitMQ是AMQP协议的Erlang的实现。
| 概念 | 说明 |
|---|---|
| 连接Connection | 一个网络连接，比如TCP/IP套接字连接。 |
| 会话Session | 端点之间的命名对话。在一个会话上下文中，保证“恰好传递一次”。 |
| 信道Channel | 多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。 |
| 客户端Client | AMQP连接或者会话的发起者。AMQP是非对称的，客户端生产和消费消息，服务器存储和路由这些消息。 |
| 服务节点Broker | 消息中间件的服务节点；一般情况下可以将一个RabbitMQ Broker看作一台RabbitMQ 服务器。 |
| 端点 | AMQP对话的任意一方。一个AMQP连接包括两个端点（一个是客户端，一个是服务器）。 |
| 消费者Consumer | 一个从消息队列里请求消息的客户端程序。 |
| 生产者Producer | 一个向交换机发布消息的客户端应用程序。 |
4.2. RabbitMQ运转流程
在入门案例中：
- 生产者发送消息
- 生产者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker；
- 声明队列并设置属性；如是否排它，是否持久化，是否自动删除；
- 将路由键（空字符串）与队列绑定起来；
- 发送消息至RabbitMQ Broker；
- 关闭信道；
- 关闭连接；
- 消费者接收消息
- 消费者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker
- 向Broker 请求消费相应队列中的消息，设置相应的回调函数；
- 等待Broker回应闭关投递响应队列中的消息，消费者接收消息；
- 确认（ack，自动确认）接收到的消息；
- RabbitMQ从队列中删除相应已经被确认的消息；
- 关闭信道；
- 关闭连接；
4.3. 生产者流转过程说明
- 客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。
- 客户端调用connection.createChannel方法。此方法开启信道，其包装的channel.open命令发送给Broker,等待channel.basicPublish方法，对应的AMQP命令为Basic.Publish,这个命令包含了content Header 和content Body()。content Header 包含了消息体的属性，例如:投递模式，优先级等，content Body 包含了消息体本身。
- 客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。
4.4. 消费者流转过程说明
- 消费者客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。
- 消费者客户端调用connection.createChannel方法。和生产者客户端一样，协议涉及Channel . Open/Open-Ok命令。
- 在真正消费之前，消费者客户端需要向Broker 发送Basic.Consume 命令(即调用channel.basicConsume 方法〉将Channel 置为接收模式，之后Broker 回执Basic . Consume - Ok 以告诉消费者客户端准备好消费消息。
- Broker 向消费者客户端推送(Push) 消息，即Basic.Deliver 命令，这个命令和Basic.Publish 命令一样会携带Content Header 和Content Body。
- 消费者接收到消息并正确消费之后，向Broker 发送确认，即Basic.Ack 命令。
- 客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。
