转发器（exchanges）：
前面的介绍都是发送者发送消息给队列，接收者从队列接收消息。下面引入Exchanges，展示RabbitMQ的完整的消息模型。
RabbitMQ消息模型的核心理念是生产者永远不会直接发送任何消息给队列，一般的情况生产者甚至不知道消息应该发送到哪些队列。相反的，生产者只能发送消息给转发器（Exchange）。转发器一边接收从生产者发来的消息，另一边把消息推送到队列中。转发器知道如何处理它收到的消息。是追加到一个指定的队列或多个队列，或者丢弃。这些规则通过转发器的类型进行定义。
前面的例子并没有使用到转发器，却仍然可以发送和接收消息。这是因为使用了一个默认的转发器，它的标识符为""，一个空字符串，称之为匿名转发器。之前发送消息的代码：
channel.basicPublish("",QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
转发器类型：Direct，Topic，Headers，Fanout。
fanout转发器发送消息：
以fanout为例，fanout转发器把所有它接收到的消息，广播到此转发器（根据channel.exchangeDeclare(EXCHANGE_NAME, "fanout" );的EXCHANGE_NAME参数区别转发器）的所有队列。
发布者：
package testrabbitmq;
import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* Created by zzhblh on 2016/8/28.
*/
public class Publisher
{
private final static String EXCHANGE_NAME = "ex_log";
public static void main(String[] args) throws IOException, TimeoutException {
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明转发器和类型
channel.exchangeDeclare(EXCHANGE_NAME, "fanout" );
String message = new Date().toLocaleString()+" : log something";
// 往转发器上发送消息
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
channel.close();
connection.close();
}
}
订阅者：
package testrabbitmq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import java.util.concurrent.TimeoutException;
/**
* Created by zzhblh on 2016/8/28.
*/
public class Subscriber
{
private final static String EXCHANGE_NAME = "ex_log";
public static void main(String[] argv) throws java.io.IOException,
java.lang.InterruptedException
{
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 创建一个非持久的、唯一的且自动删除的队列
String queueName = channel.queueDeclare().getQueue();
// 为转发器指定队列，设置binding
channel.queueBind(queueName, EXCHANGE_NAME, "");
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
// 指定接收者，第二个参数为自动应答，无需手动应答
channel.basicConsume(queueName, true, consumer);
while (true)
{
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
}
}
}
绑定（Bindings）：
上面的例子使用了绑定。类似下面的代码：channel.queueBind(queueName, EXCHANGE_NAME, "");绑定表示转发器与队列之间的关系。我们也可以简单的认为：队列对该转发器上的消息感兴趣（发布者根据EXCHANGE_NAME到转发器发送到某转发器，消费者根据EXCHANGE_NAME到此转发器上取消息）。绑定可以附带一个额外的参数routingKey。为了与避免basicPublish方法（发布消息的方法）的参数混淆，我们准备把它称作绑定键（binding key）。绑定键的意义依赖于转发器的类型。对于fanout类型，忽略此参数。
direct转发器发送消息：
direct类型的转发器背后的路由转发算法很简单：消息会被推送至绑定键（binding key）和消息发布附带的选择键（routing key）完全匹配的队列。
我们可以看到direct类型的转发器与两个队列绑定。第一个队列与绑定键orange绑定，第二个队列与转发器间有两个绑定，一个与绑定键black绑定，另一个与green绑定键绑定。
使用一个绑定键（binding key）绑定多个队列是完全合法的。如上图，一个附带选择键（routing key）的消息将会被转发到Q1和Q2。
发布者：
package testrabbitmq;
import java.io.IOException;
import java.util.Date;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* Created by zzhblh on 2016/8/28.
*/
public class NewTask
{
private static final String EXCHANGE_NAME = "ex_logs_direct";
private static final String[] SEVERITIES = { "info", "warning", "error" };
public static void main(String[] argv) throws java.io.IOException, TimeoutException {
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明转发器的类型
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
//发送6条消息
for (int i = 0; i < 6; i++)
{
String severity = getSeverity();
String message = severity + "_log :" + UUID.randomUUID().toString();
// 发布消息至转发器，指定routingkey
channel.basicPublish(EXCHANGE_NAME, severity, null, message
.getBytes());
System.out.println(" [x] Sent '" + message + "'");
}
channel.close();
connection.close();
}
/**
* 随机产生一种日志类型
*
* @return
*/
private static String getSeverity()
{
Random random = new Random();
int ranVal = random.nextInt(3);
return SEVERITIES[ranVal];
}
}
订阅者：
package testrabbitmq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import java.util.Random;
import java.util.concurrent.TimeoutException;
/**
* Created by zzhblh on 2016/8/28.
*/
public class Work
{
private static final String EXCHANGE_NAME = "ex_logs_direct";
private static final String[] SEVERITIES = { "info", "warning", "error" };
public static void main(String[] argv) throws java.io.IOException,
java.lang.InterruptedException, TimeoutException {
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明direct类型转发器
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String queueName = channel.queueDeclare().getQueue();
String severity = getSeverity();
// 指定binding_key
channel.queueBind(queueName, EXCHANGE_NAME, severity);
System.out.println(" [*] Waiting for "+severity+" logs. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while (true)
{
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
}
}
/**
* 随机产生一种日志类型
* @return
*/
private static String getSeverity()
{
Random random = new Random();
int ranVal = random.nextInt(3);
return SEVERITIES[ranVal];
}
}
发送消息时可以设置routing_key，接收队列与转发器间可以设置binding_key，接收者接收与binding_key与routing_key相同的消息。
topic转发器发送消息：
主题类型（topic）转发器的消息不能随意的设置选择键（routing_key），必须是由点隔开的一系列的标识符组成。标识符可以是任何东西，但是一般都与消息的某些特性相关。一些合法的选择键的例子："nyse.vmw","quick.orange.rabbit".你可以定义任何数量的标识符，上限为255个字节。
绑定键有两种情况。*可以匹配一个标识符。#可以匹配0个或多个标识符。我们创建3个绑定键：Q1与*.orange.*绑定Q2与*.*.rabbit和lazy.#绑定。Q1对所有的橙色动物感兴趣。Q2想要知道关于兔子的一切以及关于懒洋洋的动物的一切。一个附带quick.orange.rabbit的选择键的消息将会被转发到两个队列。附带lazy.orange.elephant的消息也会被转发到两个队列。另一方面quick.orange.fox只会被转发到Q1，lazy.brown.fox将会被转发到Q2。lazy.pink.rabbit虽然与两个绑定键匹配，但是也只会被转发到Q2一次。quick.brown.fox不能与任何绑定键匹配，所以会被丢弃。如果我们违法我们的约定，发送一个或者四个标识符的选择键，类似：orange，quick.orange.male.rabbit，这些选择键不能与任何绑定键匹配，所以消息将会被丢弃。另一方面，lazy.orange.male.rabbit，虽然是四个标识符，也可以与lazy.#匹配，从而转发至Q2。
注：主题类型的转发器非常强大，可以实现其他类型的转发器。当一个队列与绑定键#绑定，将会收到所有的消息，类似fanout类型转发器。当绑定键中不包含任何#与*时，类似direct类型转发器。
发布者：
package testrabbitmq;
import java.io.IOException;
import java.util.Date;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* Created by zzhblh on 2016/8/28.
*/
public class NewTask
{
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) throws Exception
{
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String[] routing_keys = new String[] { "kernal.info", "cron.warning",
"auth.info", "kernel.critical" };
for (String routing_key : routing_keys)
{
String msg = UUID.randomUUID().toString();
channel.basicPublish(EXCHANGE_NAME, routing_key, null, msg
.getBytes());
System.out.println(" [x] Sent routingKey = "+routing_key+" ,msg = " + msg + ".");
}
channel.close();
connection.close();
}
}
接收者：
直接收和Kernel相关的日志消息。
package testrabbitmq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import java.util.Random;
import java.util.concurrent.TimeoutException;
/**
* Created by zzhblh on 2016/8/28.
*/
public class Work
{
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) throws Exception
{
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明转发器
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
// 随机生成一个队列
String queueName = channel.queueDeclare().getQueue();
//接收所有与kernel相关的消息
channel.queueBind(queueName, EXCHANGE_NAME, "kernel.*");
System.out.println(" [*] Waiting for messages about kernel. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while (true)
{
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println(" [x] Received routingKey = " + routingKey
+ ",msg = " + message + ".");
}
}
}
Exchanges
参考：http://blog.csdn.net/lmj623565791/article/details/37607165
http://my.oschina.net/OpenSourceBO/blog/379740
