RabbitMQ/转发器/发布订阅

转发器（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