RabbitMQ是一个支持AMQP（高级消息协议）协议的消息中间件。
RabbitMQ的消息通信模型如下所示：
RabbitMQ逻辑图
消息中间件主要就是用来发送和消费消息。在RabbitMQ中，消息通过Publisher发送者将消息出去，由Consumer接收者接收到消息，并执行对应的业务逻辑。但Publisher并不是直接将消息发送给Consumer，而是先将发送到一个Exchange交换机，每一个交换机都会Binding绑定一个Queue队列，所以Publisher将消息发送到Exchange交换机后路由到对应的Queue，将消息存放在Queue中，等待Consumer来拉取数据。
消息的发布与消费简单流程如下：
- Publisher通过Channel信道将消息发送到RabbitMQ服务中的Exchange;
- Exchange通过Binding绑定规则找到对应的Queue并将消息发送到对应的Queue队列中；
- Consumer消费者通过Channel信道从Queue中拉取数据；
Exchange消息分发策略
Publisher发送消息的时候会指定Exchange交换机的名称，并指定一个路由key，该路由key会将消息发送到具体绑定了这个路由key对应的Queue中。Exchange交换机将消息发送到Queue有几种分发策略：
- direct：当指定direct分发策略时，如果消息的路由key与队列绑定的路由key相同时，交换器就会将消息发送到该队列中。例如发送消息是指定路由key为 rk1 ，那么如果队列绑定的路由key也是 rk1，那么交换机会将消息发送到该队列；
- fanout：当指定fanout分发策略时，交换机不会处理路由key，交换机会将消息发送到所有绑定了在该交换机的队列上。
- topic：当指定topic分发策略时，交换器会通过模式匹配分发消息，如果路由key与某个模式匹配时，交换机就会将消息发送到与该模式匹配的队列中。例如某个队列 queue 绑定的路由key的模式为 a.# ，当 publisher 发送消息时，如果指定发送的路由key为 a.b 或者是 a.c 时，该队列将会收到路由器发送的消息。
- headers： 这种分发策略几乎不用了。
引入pox.xml文件
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>4.1.0</version>
</dependency>
消费生产者Producer
/**
* 消息的生产者
*/
public class Producer {
public static void main(String[] args) throws Exception {
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置用户名与密码
factory.setUsername("guest");
factory.setPassword("guest");
//设置 RabbitMQ 地址
factory.setHost("localhost");
//获取连接，连接到代理服务器
Connection connection = factory.newConnection();
//获取通信信道
Channel channel = connection.createChannel();
//设置exchange交换器的名称
String exchangeName = "greeting-exchange";
//声明exchange交换器
//第二个参数声明绑定的方式为direct
//第三个参数为是否持久化,设置为true表示持久化,反之是非持久化,
// 持久化的可以将交换器存盘,在服务器重启的时候不会丢失信息
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true);
//指定路由的key
String routingKey = "xsg";
//发送的消息体
byte[] messageBodyBytes = "hello, how are you!".getBytes();
//发布信息，也就是将消息发布到greeting-exchange的交换器
// 并指定一个路由key，表示要路由到哪一个队列
channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);
//关闭信道
channel.close();
//关闭连接
connection.close();
}
}
具体步骤：
- 通过ConnectFactory获取连接工厂，设置RabbitMQ服务器的用户名、密码还有服务器地址；
- 调用ConnectFactory的newConnection方法获取一个连接Connection，连接到代理服务器；
- 通过Connection的createChannel获取到通信通道，消息将通过该通信通道发布出去；
- 定义Exchange交换机的名称，也就是要将消息发送到哪个交换机；
- 通过exchangeDeclare声明一个交换机，需要指定这个交换机分发消息的策略，这里指定为direct，表示将消息路由到路由key一致的队列中；
- 通过管道channel的basicPublish方法发布消息，需要指定交换机的名称，表示要将消息发送到哪个交换机；指定路由key，表现交换机要将消息发送到哪个队列中；
- 发送成功后关闭管道与连接；
消息的消费者Consumer
/**
* 消息的消费者
*/
public class Consumer {
public static void main(String[] args) throws Exception {
//创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置用户名与密码
factory.setUsername("guest");
factory.setPassword("guest");
//设置 RabbitMQ 地址
factory.setHost("localhost");
//获取连接，连接到代理服务器
Connection connection = factory.newConnection();
//获取通信信道
final Channel channel = connection.createChannel();
//设置exchange交换器的名称
String exchangeName = "greeting-exchange";
//声明exchange交换器
//第二个参数声明绑定的方式为direct
//第三个参数为是否持久化,设置为true表示持久化,反之是非持久化,
// 持久化的可以将交换器存盘,在服务器重启的时候不会丢失信息
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true);
//获取队列的名称
String queueName = channel.queueDeclare().getQueue();
//指定路由的key
String routingKey = "xsg";
//交换器与队列通过路由键判定起来
channel.queueBind(queueName, exchangeName, routingKey);
while (true) {
//是否自动应答
boolean autoAck = false;
String consumerTag = "";
channel.basicConsume(queueName, autoAck, consumerTag, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body) throws IOException {
String routingKey = envelope.getRoutingKey();
System.out.println("消息的路由键：" + routingKey);
//deliveryTag为条消息的一个唯一id，用来告诉RabbitMQ哪条消息被消费了
long deliveryTag = envelope.getDeliveryTag();
//第二个参数说明如何处理这个失败消息，为 true 表示该消息重新放回队列头，值为 false 表示放弃这条消息
channel.basicAck(deliveryTag, false);
System.out.println("消费的消息体内容：" + new String(body, "utf-8"));
}
});
}
}
}
具体步骤：
- 通过ConnectFactory获取连接工厂，设置RabbitMQ服务器的用户名、密码还有服务器地址；
- 调用ConnectFactory的newConnection方法获取一个连接Connection，连接到代理服务器；
- 通过Connection的createChannel获取到通信通道，消费者将通过该通信通道获取消息；
- 定义Exchange交换机的名称，也就是要将消息发送到哪个交换机；
- 通过exchangeDeclare声明一个交换机，需要指定这个交换机分发消息的策略，这里指定为direct，表示将消息路由到路由key一致的队列中；
- 通过channel.queueDeclare().getQueue() 获取队列名称，并定义一个路由key；
- 通过channel.queueBind 将交换机与队列绑定起来，需要传入队列Queue的名称、交换机的名称、路由Key，表示要将某个交换机通过路由key绑定到某个队列；
- 通过 channel.basicConsume 获取消息，需要传入队列名称，表示获取哪个队列的消息；
- 处理消息后关闭管道与连接；
运行结果：
首先运行Consumer，然后运行Product，可以看到Consumer控制台输出以下信息：
消息的路由键：xsg
消费的消息体内容：hello, how are you!
