RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等。

AMQP协议介绍

RabbitMQ 的模型架构是基于AMQP协议，生产者将消息发送给交换器，队列通过RoutingKey（路由键）绑定对应的交换机 。当生产者发送消息时携带RoutingKey，交换机根据RoutingKey找到对应的队列，将消息存入到该队列，然后消费者通过订阅该队列来获取消息。RabbitMQ 中的交换器、队列、路由键等都是遵循的 AMQP 议中相应的概念。

其中比较重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。

• 虚拟主机：本质上就是一个mini版的mq服务器，有自己的队列、交换器和绑定，最重要的，自己的权限机制。Vhost提供了逻辑上的分离，可以将众多客户端进行区分，又可以避免队列和交换器的命名冲突。Vhost必须在连接时指定，rabbitmq包含缺省vhost：“/”，通过缺省用户和口令guest进行访问。

  rabbitmq里创建用户，必须要被指派给至少一个vhost，并且只能访问被指派内的队列、交换器和绑定。Vhost必须通过rabbitmq的管理控制工具创建。

• 交换机：Exchange 用于转发消息，但是它不会做存储。这里有一个比较重要的概念：路由键 。消息到交换机的时候，交换机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
• 绑定：队列需要绑定到交换机。

常见问题

1、如果消息达到无人订阅的队列会怎么办？

消息会一直在队列中等待，RabbitMq默认队列是无限长度的。

2、多个消费者订阅到同一队列怎么办？

消息以循环的方式发送给消费者，每个消息只会发送给一个消费者。

3、消息路由到了不存在的队列怎么办？

一般情况下，凉拌，RabbitMq会忽略，当这个消息不存在，也就是这消息丢了。

Exchange类型有以下几种：

​ Fanout：广播，将消息交给所有绑定到交换机的队列

​ Direct：定向，把消息交给符合指定routing key 的队列。

​ Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

基本消息模型

每个amqp的实现都必须有一个direct交换器，包含一个空白字符串名称的默认交换器。声明一个队列时，会自动绑定到默认交换器，并且以队列名称作为路由键。

点击，可以进行查看该交换机的详情。

默认的 exchange 是一个由 broker 预创建的匿名的(即名字是空字符串) direct exchagne. 对于简单的程序来说, 默认的 exchange 有一个实用的属性: 如果没有显示地绑定 Exchnge, 那么创建的每个 queue 都会自动绑定到这个默认的 exchagne 中, 并且此时这个 queue 的 route key 就是这个queue 的名字.

下面将介绍一个例子体会一下这个最基础的消息模型

首先需要引入依赖

        <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.0.0</version></dependency>

创建一个连接的工具类ConnectionUtil，添加一个获取连接的方法

public class ConnectionUtil {/*** 建立与RabbitMQ的连接* @return* @throws Exception*/public static Connection getConnection() throws Exception {//定义连接工厂ConnectionFactory factory = new ConnectionFactory();//设置服务地址factory.setHost("182.168.6.133");//端口factory.setPort(5672);//设置账号信息，用户名、密码、vhostfactory.setVirtualHost("/hello");factory.setUsername("suzhe");factory.setPassword("suzhe");// 通过工程获取连接Connection connection = factory.newConnection();return connection;}}

生产者Producer

/*** 生产者*/
@Slf4j
public class Producer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {// 获取到连接Connection connection = ConnectionUtil.getConnection();// 创建一个信道，意味着每个线程单独一个信道Channel channel = connection.createChannel();// 声明（创建）队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 消息内容String message = "Less is more";// 向指定的队列中发送消息channel.basicPublish("", QUEUE_NAME, null, message.getBytes());log.debug("send message:{}", message);//关闭通道和连接channel.close();connection.close();}
}

运行后可以看到一条消息已经发送到了rabbitmq，并且从队里的bindings信息可以看到该队列绑定到了默认的交换机。

消费者Consumer：

/*** 消费者*/
@Slf4j
public class Consumer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {// 获取到连接Connection connection = ConnectionUtil.getConnection();// 创建一个信道，意味着每个线程单独一个信道Channel channel = connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 定义队列的消费者DefaultConsumer consumer = new DefaultConsumer(channel) {// 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,byte[] body) throws IOException {// body 即消息体String msg = new String(body);log.debug("消费消息:{}",msg);}};// 监听队列，第二个参数：是否自动进行消息确认。channel.basicConsume(QUEUE_NAME, true, consumer);}
}

运行消费者，可以看到收到了消息。

在这个例子中, 我们并没有定义 exchange, 也没有显示地将 queue 绑定到 exchange 中, 因此 queue "hello" 就自动绑定到默认的 exchange 中了, 并且在默认的 exchange 中, 其 route key 和 queue 名一致, 都为"hello"。

由于这个原因, 我们就可以使用:

        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

来发送消息。

调用 channel.basicPublish 时, 第一个参数是 exchange 名, 为空就是默认的 exchange, 第二个参数是 route key, 和 queue 名相同，第三个参数AMQP.BasicProperties 提供了一个构造器，可以通过builder() 来设置一些属性，比如

Map<String, Object> headers = new HashMap<String, Object>();headers.put("hello", "world");headers.put("aaa", "bbb");AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().deliveryMode(2) // 传送方式.contentEncoding("UTF-8") // 编码方式.expiration("10000") // 过期时间.headers(headers) //自定义属性.build();

信道

在上面的程序中，从连接中获取一个信道。

 Channel channel = connection.createChannel();

信道，概念：信道是生产消费者与rabbitmq通信的渠道，生产者publish或是消费者subscribe一个队列都是通过信道来通信的。信道是建立在TCP连接上的虚拟连接。什么意思呢？就是说rabbitmq在一条TCP上建立成百上千个信道来达到多个线程处理，这个TCP被多个线程共享，每个线程对应一个信道，信道在rabbitmq都有唯一的ID ,保证了信道私有性，对应上唯一的线程使用。

疑问：为什么不建立多个TCP连接呢？

原因是rabbitmq保证性能，系统为每个线程开辟一个TCP是非常消耗性能，每秒成百上千的建立销毁TCP会严重消耗系统。所以rabbitmq选择建立多个信道（建立在tcp的虚拟连接）连接到rabbitmq上。

本篇文章api总结：

发布消息：只用在生产者

channel.basicPublish(String exchange, //路由器的名字，即将消息发到哪个路由器String routingKey, //路由键，即发布消息时，该消息的路由键是什么BasicProperties props, //指定消息的基本属性byte[] body)//消息体，也就是消息的内容，是字节数组

BasicProperties props：指定消息的基本属性，如deliveryMode为2时表示消息持久，2以外的值表示不持久化消息

//BasicProperties介绍
String corrId = "";
String replyQueueName = "";
Integer deliveryMode = 2;
String contentType = "application/json";
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).deliveryMode(deliveryMode).contentType(contentType).build();

接收消息：只用在消费者

channel.basicConsume(String queue, //队列名字，即要从哪个队列中接收消息boolean autoAck, //是否自动确认，默认trueConsumer callback)//消费者，即谁接收消息

消费者中一般会有回调方法来消费消息

Consumer consumer = new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, //该消费者的标签Envelope envelope,//字面意思为信封：packaging data for the messageAMQP.BasicProperties properties, //message content header data byte[] body) //message bodythrows IOException {//获取消息示例String message = new String(body, "UTF-8");//接下来就可以根据消息处理一些事情}};

详细源码地址

https://github.com/suzhe2018/rabbitmq-item