RabbitMQ
一、背景
是一个由erlang开发的AMQP（Advanced Message Queue ）的开源实现。AMQP 的出现其实也是应了广大人民群众的需求，虽然在同步消息通讯的世界里有很多公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 MSMQ ，IBM 的 Websphere MQ 等），因此，在 2006 年的 6 月，Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。
二、什么是RabbitMQ？
RabbitMQ是一个在AMQP（高级消息队列协议）基础上完成的，可复用的企业消息系统，MQ全称为Message Queue，即消息队列，我们可以把它理解成一个消息中间件，或者说是消息托管服务。
三、RabbitMQ的使用场景
在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。
四、RabbitMQ的工作原理
简单的说就是一个典型的生产者消费者模型，生产者往消息队列中不断写入消息，消费者可以获取或订阅队列中的消息。
五、RabbitMQ工作流程
Queue
队列的作用是存储消息，队列的特性是先进先出。上图可以清晰地看到Client A和Client B是生产者，生产者生产消息最终被送到RabbitMQ的内部对象Queue中去，而消费者则是从Queue队列中取出数据。可以简化成表示为：
生产者Send Message “A”被传送到Queue中，消费者发现消息队列Queue中有订阅的消息，就会将这条消息A读取出来进行一些列的业务操作。这里只是一个消费正对应一个队列Queue，也可以多个消费者订阅同一个队列Queue，当然这里就会将Queue里面的消息平分给其他的消费者，但是会存在一个一个问题就是如果每个消息的处理时间不同，就会导致某些消费者一直在忙碌中，而有的消费者处理完了消息后一直处于空闲状态，因为前面已经提及到了Queue会平分这些消息给相应的消费者。这里我们就可以使用prefetchCount来限制每次发送给消费者消息的个数。详情见下图所示：
这里的prefetchCount=1是指每次从Queue中发送一条消息来。等消费者处理完这条消息后Queue会再发送一条消息给消费者。
Exchange（交换器）
首先明确一点就是生产者产生的消息并不是直接发送给消息队列Queue的，而是要经过Exchange（交换器），由Exchange再将消息路由到一个或多个Queue，当然这里还会对不符合路由规则的消息进行丢弃掉，这里指的是后续要谈到的Exchange Type。那么Exchange是怎样将消息准确的推送到对应的Queue的呢？那么这里的功劳最大的当属Binding，RabbitMQ是通过Binding将Exchange和Queue链接在一起，这样Exchange就知道如何将消息准确的推送到Queue中去。简单示意图如下所示：
在绑定（Binding）Exchange和Queue的同时，一般会指定一个Binding Key，生产者将消息发送给Exchange的时候，一般会产生一个Routing Key，当Routing Key和Binding Key对应上的时候，消息就会发送到对应的Queue中去。那么Exchange有四种类型，不同的类型有着不同的策略。也就是表明不同的类型将决定绑定的Queue不同，换言之就是说生产者发送了一个消息，Routing Key的规则是A，那么生产者会将Routing Key=A的消息推送到Exchange中，这时候Exchange中会有自己的规则，对应的规则去筛选生产者发来的消息，如果能够对应上Exchange的内部规则就将消息推送到对应的Queue中去。那么接下来就来详细讲解下Exchange里面类型。
Exchange Type
fanout:把所有发送到该Exchange的消息路由到所有与它绑定的Queue中
direct:精准匹配，Routing Key == Binding Key，当生产者（P）发送消息时Rotuing key=booking时，这时候将消息传送给Exchange，Exchange获取到生产者发送过来消息后，会根据自身的规则进行与匹配相应的Queue，这时发现Queue1和Queue2都符合，就会将消息传送给这两个队列，如果我们以Rotuing key=create和Rotuing key=confirm发送消息时，这时消息只会被推送到Queue2队列中，其他Routing Key的消息将会被丢弃
topic：简称模糊匹配，前面提到的direct规则是严格意义上的匹配，换言之Routing Key必须与Binding Key相匹配的时候才将消息传送给Queue，那么topic这个规则就是模糊匹配，可以通过通配符满足一部分规则就可以传送。它的约定是：
routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
binding key与routing key一样也是句点号“. ”分隔的字符串
binding key中可以存在两种特殊字符“*”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）
Headers：Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。
consumer（消费者）
生产者端声明了Exchange，然后绑定Queue并设置Routingkey，然后就可以往Exchange上发送消息；现在consumer怎么去拿到自己想要的消息呢？
消费者有两种方式去消费消息，一种是订阅，另一种是直接去消息服务器上直接获取；消费者端的信道中声明一个随机的消息队列，并拿到这个队列名称；然后在信道上绑定该消息队列和消息路由，对应的生产者和消费者之间都要使用相同的消息队列名称，消费者与服务器建立socket长连接进行通信，服务端收到消息推送到订阅的消费者完成消费。
六、几个重要的接口
ConnectionFactory、Connection、Channel
ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。ConnectionFactory为Connection的制造工厂。
Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。
七、安装和使用
1.安装erlang
sudo apt-get install tk tcl unixODBC erlang sudo vim /etc/profile
添加export PATH=$PATH:/usr/lib/erlang/bin/
2.安装rabbitmq
sudo apt-get install rabbitmq-serversudo添加export PATH=$PATH:/usr/lib/rabbitmq/bin
source /etc/profilerabbitmq的基本配置（端口等）参考:
3.用户与权限
在正式应用之前，我们先在RabbitMQ里创建一个vhost，加一个用户，并设置该用户的权限。使用rabbitmqctl客户端工具，在根目录下创建”/mq_test”这个vhost:
rabbitmqctl add_vhost /mq_test
创建一个用户名”test”，设置密码”test123″:
rabbitmqctl add_user test test123
设置pyh用户对/pyhtest这个vhost拥有全部权限：
rabbitmqctl set_permissions -p /mq_test test “.*” “.*” “.*”、
后面三个”*”代表pyh用户拥有对/pyhtest的配置、写、读全部权限
4.配置开启web管理插件
cat <<EOF>> /etc/rabbitmq/enabled_plugins
[rabbitmq_management].
EOF可以通过http://localhost:15672/ 查看运行情况
5.启动
使用root权限运行rabbitmq-server 或使用/etc/init.d/rabbitmq-server start|restart|stop
6.集群配置省略...
Java代码示例
生产者
package com.lieni.rabbitmq;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.lang.String;
import java.lang.System;
import java.util.HashMap;
import java.util.Map;
import java.util.Scanner;
public class Producer {
//exchange type
public enum XT {
DEFAULT, DIRECT, TOPIC, HEADERS, FANOUT
}
private static final String QUEUE_NAME = "log";
public static void main(String[] args) throws IOException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost"); //使用默认端口连接本地rabbitmq服务器
factory.setUsername("rabbitmq01");
factory.setPassword("123456");
Connection connection = factory.newConnection(); //声明一个连接
Channel channel = connection.createChannel(); //声明消息通道
//exchange类型 参考:http://stephansun.iteye.com/blog/1452853
XT xt = XT.DIRECT;
switch (xt) {
case DEFAULT: //默认，向指定的队列发送消息，消息只会被一个consumer处理,多个消费者消息会轮训处理,消息发送时如果没有consumer，消息不会丢失
//为消息通道绑定一个队列
//队列的相关参数需要与第一次定义该队列时相同，否则会出错
//参数1：队列名称
//参数2：为true时server重启队列不会消失
//参数3：队列是否是独占的，如果为true只能被一个connection使用，其他连接建立时会抛出异常
//参数4：队列不再使用时是否自动删除（没有连接，并且没有未处理的消息)
//参数5：建立队列时的其他参数
channel.queueDeclare(QUEUE_NAME, true, false, true, null);
while (GetInputString()) {
//向server发布一条消息
//参数1：exchange名字，若为空则使用默认的exchange
//参数2：routing key
//参数3：其他的属性
//参数4：消息体
//RabbitMQ默认有一个exchange，叫default exchange，它用一个空字符串表示，它是direct exchange类型，
//任何发往这个exchange的消息都会被路由到routing key的名字对应的队列上，如果没有对应的队列，则消息会被丢弃
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes()); //设置消息为持久化，服务器重启不会丢失
System.out.println("Send " + message);
}
break;
case FANOUT:
//广播给所有队列 接收方也必须通过fanout交换机获取消息,所有连接到该交换机的consumer均可获取消息
//如果producer在发布消息时没有consumer在监听，消息将被丢弃
//定义一个交换机
//参数1：交换机名称
//参数2：交换机类型
//参数3：交换机持久性，如果为true则服务器重启时不会丢失
//参数4：交换机在不被使用时是否删除
//参数5：交换机的其他属性
channel.exchangeDeclare(XCHG_NAME, "fanout", true, true, null);
while (GetInputString()) {
//发送一条广播消息,参数2此时无意义
channel.basicPublish(XCHG_NAME, "", null, message.getBytes());
System.out.println("Send " + message);
}
break;
case DIRECT:
//向所有绑定了相应routing key的队列发送消息
//如果producer在发布消息时没有consumer在监听，消息将被丢弃
//如果有多个consumer监听了相同的routing key 则他们都会受到消息
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.exchangeDeclare(XCHG_NAME, "direct", true, true, null);
channel.queueBind(QUEUE_NAME, XCHG_NAME, "info");
while (GetInputString()) {
//input like : info message
String[] temp = message.split(" ");
channel.basicPublish(XCHG_NAME, temp[0], null, temp[1].getBytes());
System.out.println("Send " + message);
}
break;
case TOPIC:
//与direct模式有类似之处，都使用routing key作为路由
//不同之处在于direct模式只能指定固定的字符串，而topic可以指定一个字符串模式
channel.exchangeDeclare(XCHG_NAME, "topic", true, true, null);
while (GetInputString()) {
//input like : topic message
String[] temp = message.split(" ");
channel.basicPublish(XCHG_NAME, temp[0], null, temp[1].getBytes());
System.out.println("Send " + message);
}
break;
case HEADERS:
//与topic和direct有一定相似之处，但不是通过routing key来路由消息
//通过headers中词的匹配来进行路由
channel.exchangeDeclare(XCHG_NAME, "headers", true, true, null);
while (GetInputString()) {
//input like : headers message
String[] temp = message.split(" ");
Map<String, Object> headers = new HashMap<String, Object>();
headers.put("name", temp[0]); //定义headers
headers.put("sex", temp[1]);
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder().headers(headers);
channel.basicPublish(XCHG_NAME, "", builder.build(), temp[2].getBytes()); //根据headers路由到相应的consumer
System.out.println("Send " + message);
}
break;
}
channel.close();
connection.close();
}
private static boolean GetInputString() {
message = scanner.nextLine();
if (message.length() == 0) return false;
return true;
}
private static Scanner scanner = new Scanner(System.in);
private static String message = "";
public static String XCHG_NAME = "xchg";
}
消费者
package com.lieni.rabbitmq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
public class Consumer {
private static final String QUEUE_NAME = "log";
public static void main(String[] args) throws IOException, InterruptedException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setUsername("rabbitmq01");
factory.setPassword("123456");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
String queueName = QUEUE_NAME;
Producer.XT xt = Producer.XT.DIRECT;
switch (xt) {
case DEFAULT:
//队列的相关参数需要与第一次定义该队列时相同，否则会出错，使用channel.queueDeclarePassive()可只被动绑定已有队列，而不创建
channel.queueDeclare(queueName, true, false, true, null);
break;
case FANOUT:
//接收端也声明一个fanout交换机
channel.exchangeDeclare(Producer.XCHG_NAME, "fanout", true, true, null);
//channel.exchangeDeclarePassive() 可以使用该函数使用一个已经建立的exchange
//声明一个临时队列，该队列会在使用完比后自动销毁
queueName = channel.queueDeclare().getQueue();
//将队列绑定到交换机,参数3无意义此时
channel.queueBind(queueName, Producer.XCHG_NAME, "");
break;
case DIRECT:
channel.queueDeclare(queueName, true, false, true, null);
channel.exchangeDeclare(Producer.XCHG_NAME, "direct", true, true, null);
//queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, Producer.XCHG_NAME, "info"); //绑定一个routing key，可以绑定多个
channel.queueBind(queueName, Producer.XCHG_NAME, "warning");
break;
case TOPIC:
channel.exchangeDeclare(Producer.XCHG_NAME, "topic", true, true, null);
queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, Producer.XCHG_NAME, "warning.#"); //监听两种模式 #匹配一个或多个单词 *匹配一个单词
channel.queueBind(queueName, Producer.XCHG_NAME, "*.blue");
break;
case HEADERS:
channel.exchangeDeclare(Producer.XCHG_NAME, "headers", true, true, null);
queueName = channel.queueDeclare().getQueue();
@SuppressWarnings("serial")
Map<String, Object> headers = new HashMap<String, Object>() {{
put("name", "test");
put("sex", "male");
put("x-match", "any");//all==匹配所有条件，any==匹配任意条件
}};
channel.queueBind(queueName, Producer.XCHG_NAME, "", headers);
break;
}
// 在同一时间不要给一个worker一个以上的消息。
// 不要将一个新的消息分发给worker知道它处理完了并且返回了前一个消息的通知标志（acknowledged）
// 替代的，消息将会分发给下一个不忙的worker。
channel.basicQos(1); //server push消息时的队列长度
//用来缓存服务器推送过来的消息
QueueingConsumer consumer = new QueueingConsumer(channel);
//为channel声明一个consumer，服务器会推送消息
//参数1:队列名称
//参数2：是否发送ack包，不发送ack消息会持续在服务端保存，直到收到ack。 可以通过channel.basicAck手动回复ack
//参数3：消费者
channel.basicConsume(queueName, false, consumer);
//channel.basicGet(queueName, true); //使用该函数主动去服务器检索是否有新消息，而不是等待服务器推送
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
System.out.println("Received " + new String(delivery.getBody()));
//回复ack包，如果不回复，消息不会在服务器删除
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
//channel.basicReject(); channel.basicNack(); //可以通过这两个函数拒绝消息，可以指定消息在服务器删除还是继续投递给其他消费者
}
}
}