RabbitMQ入门

[toc]

一:入门

1.安装Erlang

2.安装RabbitMQ

3.配置

激活 RabbitMQ's Management Plugin

1C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.12\sbin>rabbitmq-plugins.bat enable rabbitmq
2Error: The following plugins could not be found:
3  rabbitmq
4
5
6C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.12\sbin>rabbitmq-plugins.bat enable rabbitmq_management
7The following plugins have been enabled:
8  amqp_client
9  cowlib
10  cowboy
11  rabbitmq_web_dispatch
12  rabbitmq_management_agent
13  rabbitmq_management
14
15Applying plugin configuration to rabbit@DESKTOP-BugYang... started 6 plugins.
16
17C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.12\sbin>net stop RabbitMQ && net start RabbitMQ
18RabbitMQ 服务正在停止......
19RabbitMQ 服务已成功停止。
20
21RabbitMQ 服务正在启动 .
22RabbitMQ 服务已经启动成功。
23
24
25C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.12\sbin>rabbitmqctl.bat list_users
26Listing users
27guest   [administrator]
28
29C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.12\sbin>
30
31http://localhost:15672/

账号密码:guest

4.下载maven

1<dependency>
2  <groupId>com.rabbitmq</groupId>
3  <artifactId>amqp-client</artifactId>
4  <version>4.1.0</version>
5</dependency>

5.创建发送者

1public class Send {
2
3    //队列名称
4    private final static String QUEUE_NAME = "hello";
5
6    public static void main(String[] argv) throws java.io.IOException, TimeoutException {
7        /**
8         * 创建连接连接到MabbitMQ
9         */
10        ConnectionFactory factory = new ConnectionFactory();
11        //设置MabbitMQ所在主机ip或者主机名
12        factory.setHost("localhost");
13        //创建一个连接
14        Connection connection = factory.newConnection();
15        //创建一个频道
16        Channel channel = connection.createChannel();
17        //指定一个队列
18        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
19        //发送的消息
20        String message = "hello world!";
21        //往队列中发出一条消息
22        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
23        System.out.println(" [x] Sent '" + message + "'");
24        //关闭频道和连接
25        channel.close();
26        connection.close();
27    }
28}

打印

1SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
2SLF4J: Defaulting to no-operation (NOP) logger implementation
3SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
4 [x] Sent 'hello world!'
5
6Process finished with exit code 0

6.创建接受者

1public class Rec {
2    //队列名称
3    private final static String QUEUE_NAME = "hello";
4
5    public static void main(String[] argv) throws java.io.IOException,
6            java.lang.InterruptedException, TimeoutException {
7        //打开连接和创建频道，与发送端一样
8        ConnectionFactory factory = new ConnectionFactory();
9        factory.setHost("localhost");
10        Connection connection = factory.newConnection();
11        Channel channel = connection.createChannel();
12        //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
13        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
14        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
15
16        //创建队列消费者
17        QueueingConsumer consumer = new QueueingConsumer(channel);
18        //指定消费队列
19        channel.basicConsume(QUEUE_NAME, true, consumer);
20        while (true)
21        {
22            //nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
23            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
24            String message = new String(delivery.getBody());
25            System.out.println(" [x] Received '" + message + "'");
26        }
27
28    }
29}

打印

1SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
2SLF4J: Defaulting to no-operation (NOP) logger implementation
3SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
4 [*] Waiting for messages. To exit press CTRL+C
5 [x] Received 'hello world!'

二:工作队列

1.发送消息

1public class NewTask
2{
3    //队列名称
4    private final static String QUEUE_NAME = "workqueue";
5
6    public static void main(String[] args) throws IOException, TimeoutException {
7
8        //创建连工厂
9        ConnectionFactory factory = new ConnectionFactory();
10        //设置ip
11        factory.setHost("localhost");
12        //创建连接
13        Connection connection = factory.newConnection();
14        //创建队列
15        Channel channel = connection.createChannel();
16        //声明队列
17        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
18        //发送10条消息，依次在消息后面附加1-10个点
19        for (int i = 0; i < 10; i++)
20        {
21            String dots = "";
22            for (int j = 0; j <= i; j++)
23            {
24                dots += ".";
25            }
26            //拼数据
27            String message = "helloworld" + dots+dots.length();
28            //推送到rabbitmq中
29            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
30            //推送完成,打印结束语句
31            System.out.println(" [x] Sent '" + message + "'");
32        }
33        //关闭队列
34        channel.close();
35        //关闭消息
36        connection.close();
37
38    }
39
40
41}
42
43 [x] Sent 'helloworld.1'
44 [x] Sent 'helloworld..2'
45 [x] Sent 'helloworld...3'
46 [x] Sent 'helloworld....4'
47 [x] Sent 'helloworld.....5'
48 [x] Sent 'helloworld......6'
49 [x] Sent 'helloworld.......7'
50 [x] Sent 'helloworld........8'
51 [x] Sent 'helloworld.........9'
52 [x] Sent 'helloworld..........10'

2.接收消息

运行两个Work类

1public class Work
2{
3    //队列名称
4    private final static String QUEUE_NAME = "workqueue";
5
6    public static void main(String[] argv) throws java.io.IOException,
7            java.lang.InterruptedException, TimeoutException {
8
9        //区分不同工作进程的输出
10        int hashCode = Work.class.hashCode();
11
12        //创建连接工厂
13        ConnectionFactory factory = new ConnectionFactory();
14        //设置ip
15        factory.setHost("localhost");
16        //创建连接
17        Connection connection = factory.newConnection();
18        //创建队列
19        Channel channel = connection.createChannel();
20        //声明队列
21        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
22        System.out.println(hashCode
23                + " [*] Waiting for messages. To exit press CTRL+C");
24
25        QueueingConsumer consumer = new QueueingConsumer(channel);
26       // 指定消费队列
27        //关闭应答机制,会丢失消息
28        channel.basicConsume(QUEUE_NAME, true, consumer);
29        //打开应答机制,不会丢失消息
30        channel.basicConsume(QUEUE_NAME, false, consumer);
31        while (true)
32        {
33            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
34            String message = new String(delivery.getBody());
35
36            System.out.println(hashCode + " [x] Received '" + message + "'");
37//            doWork(message);
38            System.out.println(hashCode + " [x] Done");
39
40        }
41
42    }
43
44    /**
45     * 每个点耗时1s
46     * @param task
47     * @throws InterruptedException
48     */
49    private static void doWork(String task) throws InterruptedException
50    {
51        for (char ch : task.toCharArray())
52        {
53            if (ch == '.')
54                Thread.sleep(1000);
55        }
56    }
57}
58
59746292446 [x] Received 'helloworld.1'
60746292446 [x] Done
61746292446 [x] Received 'helloworld...3'
62746292446 [x] Done
63746292446 [x] Received 'helloworld.....5'
64746292446 [x] Done
65746292446 [x] Received 'helloworld.......7'
66746292446 [x] Done
67746292446 [x] Received 'helloworld.........9'
68746292446 [x] Done
69
70242131142 [x] Received 'helloworld..2'
71242131142 [x] Done
72242131142 [x] Received 'helloworld....4'
73242131142 [x] Done
74242131142 [x] Received 'helloworld......6'
75242131142 [x] Done
76242131142 [x] Received 'helloworld........8'
77242131142 [x] Done
78242131142 [x] Received 'helloworld..........10'
79242131142 [x] Done

可以看到，默认的，RabbitMQ会一个一个的发送信息给下一个消费者(consumer)，而不考虑每个任务的时长等等，且是一次性分配，并非一个一个分配。平均的每个消费者将会获得相等数量的消息。这样分发消息的方式叫做round-robin。

3.消息应答（message acknowledgments）

我们首先开启两个任务，然后执行发送任务的代码（NewTask.java），然后立即关闭第二个任务,两个加起来打印出来的数据会有缺失

一旦RabbItMQ交付了一个信息给消费者，会马上从内存中移除这个信息。在这种情况下，如果杀死正在执行任务的某个工作者，我们会丢失它正在处理的信息。我们也会丢失已经转发给这个工作者且它还未执行的消息。

为了保证消息永远不会丢失，RabbitMQ支持消息应答（message acknowledgments）。

消费者发送应答给RabbitMQ，告诉它信息已经被接收和处理，然后RabbitMQ可以自由的进行信息删除。
如果消费者被杀死而没有发送应答，RabbitMQ会认为该信息没有被完全的处理，然后将会重新转发给别的消费者。通过这种方式，你可以确认信息不会被丢失，即使消者偶尔被杀死。
这种机制并没有超时时间这么一说，RabbitMQ只有在消费者连接断开是重新转发此信息。如果消费者处理一个信息需要耗费特别特别长的时间是允许的。

消息应答默认是打开的。上面的代码中我们通过显示的设置autoAsk=true关闭了这种机制。

1boolean ack = false ; //打开应答机制  
2channel.basicConsume(QUEUE_NAME, ack, consumer);  
3//另外需要在每次处理完成一个消息后，手动发送一次应答。  
4channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
5
6public class Work  
7{  
8    //队列名称  
9    private final static String QUEUE_NAME = "workqueue";  
10  
11    public static void main(String[] argv) throws java.io.IOException,  
12            java.lang.InterruptedException  
13    {  
14        //区分不同工作进程的输出  
15        int hashCode = Work.class.hashCode();  
16        //创建连接和频道  
17        ConnectionFactory factory = new ConnectionFactory();  
18        factory.setHost("localhost");  
19        Connection connection = factory.newConnection();  
20        Channel channel = connection.createChannel();  
21        //声明队列  
22        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
23        System.out.println(hashCode  
24                + " [*] Waiting for messages. To exit press CTRL+C");  
25        QueueingConsumer consumer = new QueueingConsumer(channel);  
26        // 指定消费队列  
27        boolean ack = false ; //打开应答机制  
28        channel.basicConsume(QUEUE_NAME, ack, consumer);  
29        while (true)  
30        {  
31            QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
32            String message = new String(delivery.getBody());  
33  
34            System.out.println(hashCode + " [x] Received '" + message + "'");  
35            doWork(message);  
36            System.out.println(hashCode + " [x] Done");  
37            //发送应答  
38            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
39  
40        }  
41  
42    }  
43}

4.消息持久化（Message durability）

我们已经学习了即使消费者被杀死，消息也不会被丢失。但是如果此时RabbitMQ服务被停止，我们的消息仍然会丢失

当RabbitMQ退出或者异常退出，将会丢失所有的队列和信息，除非你告诉它不要丢失。

我们需要做两件事来确保信息不会被丢失：我们需要给所有的队列和消息设置持久化的标志。

第一，我们需要确认RabbitMQ永远不会丢失我们的队列。为了这样，我们需要声明它为持久化的。

boolean durable = true; channel.queueDeclare("task_queue", durable, false, false, null);

注：RabbitMQ不允许使用不同的参数重新定义一个队列，所以已经存在的队列，我们无法修改其属性。

第二，我们需要标识我们的信息为持久化的。通过设置MessageProperties（implements BasicProperties）值为PERSISTENT_TEXT_PLAIN。

channel.basicPublish("", "task_queue",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

现在你可以执行一个发送消息的程序，然后关闭服务，再重新启动服务，运行消费者程序做下实验。

5.公平转发（Fair dispatch）

对于两个消费者，有一系列的任务，奇数任务特别耗时，而偶数任务却很轻松，这样造成一个消费者一直繁忙，另一个消费者却很快执行完任务后等待。
造成这样的原因是因为RabbitMQ仅仅是当消息到达队列进行转发消息。并不在乎有多少任务消费者并未传递一个应答给RabbitMQ。仅仅盲目转发所有的奇数给一个消费者，偶数给另一个消费者。

1int prefetchCount = 1;  
2channel.basicQos(prefetchCount);
3
4public class NewTask  
5{  
6    // 队列名称  
7    private final static String QUEUE_NAME = "workqueue_persistence";  
8  
9    public static void main(String[] args) throws IOException  
10    {  
11        // 创建连接和频道  
12        ConnectionFactory factory = new ConnectionFactory();  
13        factory.setHost("localhost");  
14        Connection connection = factory.newConnection();  
15        Channel channel = connection.createChannel();  
16        // 声明队列  
17        boolean durable = true;// 1、设置队列持久化  
18        channel.queueDeclare(QUEUE_NAME, durable, false, false, null);  
19        // 发送10条消息，依次在消息后面附加1-10个点  
20        for (int i = 5; i > 0; i--)  
21        {  
22            String dots = "";  
23            for (int j = 0; j <= i; j++)  
24            {  
25                dots += ".";  
26            }  
27            String message = "helloworld" + dots + dots.length();  
28            // MessageProperties 2、设置消息持久化  
29            channel.basicPublish("", QUEUE_NAME,  
30                    MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());  
31            System.out.println(" [x] Sent '" + message + "'");  
32        }  
33        // 关闭频道和资源  
34        channel.close();  
35        connection.close();  
36  
37    }  
38  
39}
40
41public class Work  
42{  
43    // 队列名称  
44    private final static String QUEUE_NAME = "workqueue_persistence";  
45  
46    public static void main(String[] argv) throws java.io.IOException,  
47            java.lang.InterruptedException  
48    {  
49        // 区分不同工作进程的输出  
50        int hashCode = Work.class.hashCode();  
51        // 创建连接和频道  
52        ConnectionFactory factory = new ConnectionFactory();  
53        factory.setHost("localhost");  
54        Connection connection = factory.newConnection();  
55        Channel channel = connection.createChannel();  
56        // 声明队列  
57        boolean durable = true;  
58        channel.queueDeclare(QUEUE_NAME, durable, false, false, null);  
59        System.out.println(hashCode  
60                + " [*] Waiting for messages. To exit press CTRL+C");  
61        //设置最大服务转发消息数量  
62        int prefetchCount = 1;  
63        channel.basicQos(prefetchCount);  
64        QueueingConsumer consumer = new QueueingConsumer(channel);  
65        // 指定消费队列  
66        boolean ack = false; // 打开应答机制  
67        channel.basicConsume(QUEUE_NAME, ack, consumer);  
68        while (true)  
69        {  
70            QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
71            String message = new String(delivery.getBody());  
72  
73            System.out.println(hashCode + " [x] Received '" + message + "'");  
74            doWork(message);  
75            System.out.println(hashCode + " [x] Done");  
76            //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
77            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
78  
79        }  
80  
81    }  
82  
83    /** 
84     * 每个点耗时1s 
85     *  
86     * @param task 
87     * @throws InterruptedException 
88     */  
89    private static void doWork(String task) throws InterruptedException  
90    {  
91        for (char ch : task.toCharArray())  
92        {  
93            if (ch == '.')  
94                Thread.sleep(1000);  
95        }  
96    }  
97}

三:发布/订阅

工作队列中的一个任务只会发给一个工作者

就是把一个消息发给多个消费者，这种模式称之为发布/订阅（类似观察者模式）。

为了验证这种模式，我们准备构建一个简单的日志系统。这个系统包含两类程序，

一类程序发动日志，另一类程序接收和处理日志。

我们实现，一个接收者将接收到的数据写到硬盘上，与此同时，另一个接收者把接收到的消息展现在屏幕上。

1:转发器（Exchanges）

RabbitMQ消息模型的核心理念是生产者永远不会直接发送任何消息给队列，一般的情况生产者甚至不知道消息应该发送到哪些队列。

相反的，生产者只能发送消息给转发器（Exchange）。转发器是非常简单的，一边接收从生产者发来的消息，另一边把消息推送到队列中。转发器必须清楚的知道消息如何处理它收到的每一条消息。是否应该追加到一个指定的队列？是否应该追加到多个队列？或者是否应该丢弃？这些规则通过转发器的类型进行定义。

可用的转发器类型：

Direct
Topic
Headers
Fanout

声明转发器类型的代码：

channel.exchangeDeclare("logs","fanout");

fanout类型转发器特别简单，把所有它介绍到的消息，广播到所有它所知道的队列。不过这正是我们前述的日志系统所需要的

2、匿名转发器（nameless exchange）

前面说到生产者只能发送消息给转发器（Exchange），但是我们前两篇博客中的例子并没有使用到转发器，我们仍然可以发送和接收消息。

这是因为我们使用了一个默认的转发器，它的标识符为””。之前发送消息的代码：

channel.basicPublish("", QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

第一个参数为转发器的名称，我们设置为”” : 如果存在routingKey（第二个参数），消息由routingKey决定发送到哪个队列。

现在我们可以指定消息发送到的转发器：

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

3、临时队列（Temporary queues）

前面的博客中我们都为队列指定了一个特定的名称。能够为队列命名对我们来说是很关键的，我们需要指定消费者为某个队列。当我们希望在生产者和消费者间共享队列时，为队列命名是很重要的。
不过，对于我们的日志系统我们并不关心队列的名称。我们想要接收到所有的消息，而且我们也只对当前正在传递的数据的感兴趣。为了满足我们的需求，需要做两件事：
第一，无论什么时间连接到Rabbit我们都需要一个新的空的队列。为了实现，我们可以使用随机数创建队列，或者更好的，让服务器给我们提供一个随机的名称。
第二，一旦消费者与Rabbit断开，消费者所接收的那个队列应该被自动删除。
Java中我们可以使用queueDeclare()方法，不传递任何参数，来创建一个非持久的、唯一的、自动删除的队列且队列名称由服务器随机产生。

String queueName = channel.queueDeclare().getQueue();

一般情况这个名称与amq.gen-JzTY20BRgKO-HjmUJj0wLg 类似

4、绑定（Bindings）

我们已经创建了一个fanout转发器和队列，我们现在需要通过binding告诉转发器把消息发送给我们的队列。

channel.queueBind(queueName, “logs”, ””)

参数1：队列名称；参数2：转发器名称

5、完整的例子

1.创建发送器

1public class EmitLog
2{
3    private final static String EXCHANGE_NAME = "ex_log";
4
5    public static void main(String[] args) throws IOException, TimeoutException {
6        // 创建连接工厂
7        ConnectionFactory factory = new ConnectionFactory();
8        //设置ip
9        factory.setHost("localhost");
10        //创建连接
11        Connection connection = factory.newConnection();
12        //创建频道
13        Channel channel = connection.createChannel();
14        // 声明转发器和类型
15        channel.exchangeDeclare(EXCHANGE_NAME, "fanout" );
16        //创建发送的数据
17        String message = new Date().toLocaleString()+" : log something";
18        // 往转发器上发送消息
19        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
20
21        System.out.println(" [x] Sent '" + message + "'");
22
23        channel.close();
24        connection.close();
25
26    }
27
28}

2.创建接收器,数据写进文件里

1public class ReceiveLogsToSave
2{
3    private final static String EXCHANGE_NAME = "ex_log";
4
5    public static void main(String[] argv) throws java.io.IOException,
6            java.lang.InterruptedException, TimeoutException {
7
8        // 创建连接工厂
9        ConnectionFactory factory = new ConnectionFactory();
10        //设置ip
11        factory.setHost("localhost");
12        //创建连接
13        Connection connection = factory.newConnection();
14        //创建频道
15        Channel channel = connection.createChannel();
16        // 声明转发器和类型
17        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
18        // 创建一个非持久的、唯一的且自动删除的队列,临时队列
19        String queueName = channel.queueDeclare().getQueue();
20        // 为转发器指定队列，设置binding,绑定
21        channel.queueBind(queueName, EXCHANGE_NAME, "");
22
23        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
24
25        QueueingConsumer consumer = new QueueingConsumer(channel);
26        // 指定接收者，第二个参数为自动应答，无需手动应答
27        channel.basicConsume(queueName, true, consumer);
28
29        while (true)
30        {
31            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
32            String message = new String(delivery.getBody());
33
34            print2File(message);
35        }
36
37    }
38
39    private static void print2File(String msg)
40    {
41        try
42        {
43            String dir = ReceiveLogsToSave.class.getClassLoader().getResource("").getPath();
44            String logFileName = new SimpleDateFormat("yyyy-MM-dd")
45                    .format(new Date());
46            File file = new File(dir, logFileName+".txt");
47            FileOutputStream fos = new FileOutputStream(file, true);
48            fos.write((msg + "\r\n").getBytes());
49            fos.flush();
50            fos.close();
51        } catch (FileNotFoundException e)
52        {
53            e.printStackTrace();
54        } catch (IOException e)
55        {
56            e.printStackTrace();
57        }
58    }
59}

3.创建接收器,打印出信息

1public class ReceiveLogsToConsole
2{
3    private final static String EXCHANGE_NAME = "ex_log";
4
5    public static void main(String[] argv) throws java.io.IOException,
6            java.lang.InterruptedException, TimeoutException {
7        // 创建连接工厂
8        ConnectionFactory factory = new ConnectionFactory();
9        //设置ip
10        factory.setHost("localhost");
11        //创建连接
12        Connection connection = factory.newConnection();
13        //创建频道
14        Channel channel = connection.createChannel();
15        // 声明转发器和类型
16        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
17        // 创建一个非持久的、唯一的且自动删除的队列
18        String queueName = channel.queueDeclare().getQueue();
19        // 为转发器指定队列，设置binding
20        channel.queueBind(queueName, EXCHANGE_NAME, "");
21
22        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
23
24        QueueingConsumer consumer = new QueueingConsumer(channel);
25        // 指定接收者，第二个参数为自动应答，无需手动应答
26        channel.basicConsume(queueName, true, consumer);
27
28        while (true)
29        {
30            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
31            String message = new String(delivery.getBody());
32            System.out.println(" [x] Received '" + message + "'");
33
34        }
35
36    }
37
38}

四:路由(Routing)

需求:本篇博客我们准备给日志系统添加新的特性，让日志接收者能够订阅部分消息。例如，我们可以仅仅将致命的错误写入日志文件，然而仍然在控制面板上打印出所有的其他类型的日志消息。

1、绑定（Bindings）

在上一篇博客中我们已经使用过绑定

channel.queueBind(queueName, EXCHANGE_NAME, "");

绑定表示转发器与队列之间的关系。

我们也可以简单的认为：队列对该转发器上的消息感兴趣。

绑定可以附带一个额外的参数routingKey。为了与避免basicPublish方法（发布消息的方法）的参数混淆，我们准备把它称作绑定键（binding key）。下面展示如何使用绑定键（binding key）来创建一个绑定：

channel.queueBind(queueName, EXCHANGE_NAME, "black");

绑定键的意义依赖于转发器的类型。对于fanout类型，忽略此参数。

2、直接转发（Direct exchange）

上一篇的日志系统广播所有的消息给所有的消费者。

现在想:可能希望把致命类型的错误写入硬盘，而不把硬盘空间浪费在警告或者消息类型的日志上。

之前我们使用fanout类型的转发器，但是并没有给我们带来更多的灵活性：仅仅可以愚蠢的转发。

我们将会使用direct类型的转发器进行替代。direct类型的转发器背后的路由转发算法很简单：

消息会被推送至绑定键（binding key）和消息发布附带的选择键（routing key）完全匹配的队列。

图解：

img

我们可以看到direct类型的转发器与两个队列绑定。

第一个队列与绑定键orange绑定

第二个队列与转发器间有两个绑定，一个与绑定键black绑定，另一个与green绑定键绑定。
这样的话，当一个消息附带一个选择键（routing key） orange发布至转发器将会被导向到队列Q1。消息附带一个选择键（routing key）black或者green将会被导向到Q2.所有的其他的消息将会被丢弃。

3、多重绑定（multiple bindings）

img

使用一个绑定键（binding key）绑定多个队列是完全合法的。如上图，一个附带选择键（routing key）的消息将会被转发到Q1和Q2。

4、发送日志（Emittinglogs）

我们将消息发送到direct类型的转发器而不是fanout类型。我们将把日志的严重性作为选择键（routing key）。这样的话，接收程序可以根据严重性来选择接收。我们首先关注发送日志的代码：

像以前一样，我们需要先创建一个转发器：

channel.exchangeDeclare(EXCHANGE_NAME,"direct");

然后我们准备发送一条消息：

channel.basicPublish(EXCHANGE_NAME,severity, null, message.getBytes());

为了简化代码，我们假定‘severity’是‘info’，‘warning’，‘error’中的一个。

5、订阅

接收消息的代码和前面的博客的中类似，只有一点不同：我们给我们所感兴趣的严重性类型的日志创建一个绑定。

1StringqueueName = channel.queueDeclare().getQueue();
2for(Stringseverity : argv){
3    channel.queueBind(queueName, EXCHANGE_NAME, severity);
4}

6、完整的实例

1.创建发送者

1public class EmitLogDirect
2{
3
4    private static final String EXCHANGE_NAME = "ex_logs_direct";
5    private static final String[] SEVERITIES = { "info", "warning", "error" };
6
7    public static void main(String[] argv) throws java.io.IOException, TimeoutException {
8        // 创建连接工厂
9        ConnectionFactory factory = new ConnectionFactory();
10        //设置ip
11        factory.setHost("localhost");
12        //创建连接
13        Connection connection = factory.newConnection();
14        //创建频道
15        Channel channel = connection.createChannel();
16        // 声明转发器的类型
17        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
18
19        //发送6条消息
20        for (int i = 0; i < 6; i++)
21        {
22            String severity = getSeverity();
23            String message = severity + "_log :" + UUID.randomUUID().toString();
24            // 发布消息至转发器，指定routingkey
25            channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
26            System.out.println(" [x] Sent '" + message + "'");
27        }
28
29        channel.close();
30        connection.close();
31    }
32
33    /**
34     * 随机产生一种日志类型
35     *
36     * @return
37     */
38    private static String getSeverity()
39    {
40        Random random = new Random();
41        int ranVal = random.nextInt(3);
42        return SEVERITIES[ranVal];
43    }
44}

2.创建接受者

1public class ReceiveLogsDirect
2{
3
4    private static final String EXCHANGE_NAME = "ex_logs_direct";
5    private static final String[] SEVERITIES = { "info", "warning", "error" };
6
7    public static void main(String[] argv) throws java.io.IOException,
8            java.lang.InterruptedException, TimeoutException {
9        // 创建连接工厂
10        ConnectionFactory factory = new ConnectionFactory();
11        //设置ip
12        factory.setHost("localhost");
13        //创建连接
14        Connection connection = factory.newConnection();
15        //创建频道
16        Channel channel = connection.createChannel();
17        // 声明direct类型转发器
18        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
19        // 创建一个非持久的、唯一的且自动删除的队列,临时队列
20        String queueName = channel.queueDeclare().getQueue();
21
22        String severity = getSeverity();
23
24        // 指定binding_key
25        channel.queueBind(queueName, EXCHANGE_NAME, severity);
26        System.out.println(" [*] Waiting for "+severity+" logs. To exit press CTRL+C");
27
28        QueueingConsumer consumer = new QueueingConsumer(channel);
29        channel.basicConsume(queueName, true, consumer);
30
31        while (true)
32        {
33            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
34            String message = new String(delivery.getBody());
35
36            System.out.println(" [x] Received '" + message + "'");
37        }
38    }
39
40    /**
41     * 随机产生一种日志类型
42     *
43     * @return
44     */
45    private static String getSeverity()
46    {
47        Random random = new Random();
48        int ranVal = random.nextInt(3);
49        return SEVERITIES[ranVal];
50    }
51}

3.总结:

发送消息时可以设置routing_key，接收队列与转发器间可以设置binding_key，接收者接收与binding_key与routing_key相同的消息。

五:主题

1、主题转发（Topic Exchange）

发往主题类型的转发器的消息不能随意的设置选择键（routing_key），必须是由点隔开的一系列的标识符组成。标识符可以是任何东西，但是一般都与消息的某些特性相关。一些合法的选择键的例子："stock.usd.nyse", "nyse.vmw","quick.orange.rabbit".你可以定义任何数量的标识符，上限为255个字节。

2. 3.完整例子

1.发送

1public class EmitLogTopic
2{
3
4    private static final String EXCHANGE_NAME = "topic_logs";
5
6    public static void main(String[] argv) throws Exception
7    {
8        // 创建连接和频道
9        ConnectionFactory factory = new ConnectionFactory();
10        factory.setHost("localhost");
11        Connection connection = factory.newConnection();
12        Channel channel = connection.createChannel();
13
14        //指定topic的转发器
15        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
16
17        String[] routing_keys = new String[] { "kernal.info", "cron.warning","auth.info", "kernel.critical" };
18
19        for (String routing_key : routing_keys)
20        {
21            String msg = UUID.randomUUID().toString();
22            channel.basicPublish(EXCHANGE_NAME, routing_key, null, msg.getBytes());
23            System.out.println(" [x] Sent routingKey = "+routing_key+" ,msg = " + msg + ".");
24        }
25
26        channel.close();
27        connection.close();
28    }
29}

2.接收1

1public class ReceiveLogsTopicForCritical
2{
3
4    private static final String EXCHANGE_NAME = "topic_logs";
5
6    public static void main(String[] argv) throws Exception
7    {
8        // 创建连接和频道
9        ConnectionFactory factory = new ConnectionFactory();
10        factory.setHost("localhost");
11        Connection connection = factory.newConnection();
12        Channel channel = connection.createChannel();
13        // 声明topic转发器
14        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
15        // 随机生成一个队列
16        String queueName = channel.queueDeclare().getQueue();
17
18        // 接收所有与kernel相关的消息
19        channel.queueBind(queueName, EXCHANGE_NAME, "*.critical");
20
21        QueueingConsumer consumer = new QueueingConsumer(channel);
22        channel.basicConsume(queueName, true, consumer);
23
24        while (true)
25        {
26            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
27            String message = new String(delivery.getBody());
28            String routingKey = delivery.getEnvelope().getRoutingKey();
29
30            System.out.println(" [x] Received routingKey = " + routingKey  + ",msg = " + message + ".");
31        }
32    }
33}

2.接收2

1public class ReceiveLogsTopicForKernel
2{
3
4    private static final String EXCHANGE_NAME = "topic_logs";
5
6    public static void main(String[] argv) throws Exception
7    {
8        // 创建连接和频道
9        ConnectionFactory factory = new ConnectionFactory();
10        factory.setHost("localhost");
11        Connection connection = factory.newConnection();
12        Channel channel = connection.createChannel();
13        // 声明topic转发器
14        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
15        // 随机生成一个队列
16        String queueName = channel.queueDeclare().getQueue();
17
18        //接收所有与kernel相关的消息
19        channel.queueBind(queueName, EXCHANGE_NAME, "kernel.*");
20
21        QueueingConsumer consumer = new QueueingConsumer(channel);
22        channel.basicConsume(queueName, true, consumer);
23
24        while (true)
25        {
26            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
27            String message = new String(delivery.getBody());
28            String routingKey = delivery.getEnvelope().getRoutingKey();
29
30            System.out.println(" [x] Received routingKey = " + routingKey + ",msg = " + message + ".");
31        }
32    }
33}

六:参考

http://blog.csdn.net/lmj623565791/article/details/37706355

http://blog.csdn.net/lmj623565791/article/details/37669573

http://blog.csdn.net/lmj623565791/article/details/37657225

http://blog.csdn.net/lmj623565791/article/details/37620057

http://blog.csdn.net/lmj623565791/article/details/37607165