RabbitMQ学习笔记3-使用topic交换器

 1 package com.yzl.test2;
 2 
 3 import java.util.concurrent.CountDownLatch;
 4 import java.util.concurrent.ExecutorService;
 5 import java.util.concurrent.Executors;
 6 
 7 import com.rabbitmq.client.Channel;
 8 import com.rabbitmq.client.Connection;
 9 import com.rabbitmq.client.ConnectionFactory;
10 
11 /**
12  * 使用topic交换器发送消息
13  * 分为警告和错误2种日志
14  * @author: yzl
15  * @date: 2016-10-22
16  */
17 public class SenderWithTopicExchange {
18     //交换器名称
19     private static final String EXCHANGE_NAME = "myTopicExchange";
20     //路由键的前缀
21     private static final String BASE_ROUTING_KEY = "logger.";
22     
23     public static void main(String[] args) throws Exception {
24         //使用CountDownLatch控制2个线程一起运行
25         final CountDownLatch cdl = new CountDownLatch(2);
26         //连接到rabbitmq服务器
27         ConnectionFactory factory = new ConnectionFactory();
28         factory.setHost("localhost");
29         Connection connection = factory.newConnection();
30         //创建一个信道
31         final Channel channel = connection.createChannel();
32         //定义一个名字为topicExchange的topic类型的exchange
33         channel.exchangeDeclare(EXCHANGE_NAME, "topic");
34         
35         ExecutorService pool = Executors.newFixedThreadPool(2);
36         pool.submit(new Runnable() {
37             @Override
38             public void run() {
39                 try {
40                     cdl.await();
41                     //发送警告日志,绑定路由键：logger.warning
42                     String warningMsg = "warning message is :";
43                     for(int i=1; i<800; i++){
44                         System.out.println("发送警告消息：" + warningMsg+i);
45                         channel.basicPublish(EXCHANGE_NAME, BASE_ROUTING_KEY + "warning", null, (warningMsg+i).getBytes());
46                         Thread.sleep(2000L);
47                     }
48                 } catch (Exception e) {
49                     e.printStackTrace();
50                 }
51             }
52         });
53         pool.submit(new Runnable() {
54             @Override
55             public void run() {
56                 try {
57                     cdl.await();
58                     //发送错误日志,绑定路由键：logger.error
59                     String errorMsg = "error message is :";
60                     for(int i=1; i<1000; i++){
61                         System.out.println("发送错误消息：" + errorMsg+i);
62                         channel.basicPublish(EXCHANGE_NAME, BASE_ROUTING_KEY + "error", null, (errorMsg+i).getBytes());
63                         Thread.sleep(2000L);
64                     }
65                 } catch (Exception e) {
66                     e.printStackTrace();
67                 }
68             }
69         });
70         
71         cdl.countDown();
72         cdl.countDown();
73     }
74 }

 1 package com.yzl.test2;
 2 
 3 import java.io.IOException;
 4 
 5 import com.rabbitmq.client.AMQP.BasicProperties;
 6 import com.rabbitmq.client.Channel;
 7 import com.rabbitmq.client.Connection;
 8 import com.rabbitmq.client.ConnectionFactory;
 9 import com.rabbitmq.client.DefaultConsumer;
10 import com.rabbitmq.client.Envelope;
11 
12 /**
13  * 使用topic交换器接收消息
14  * 
15  * @author: yzl
16  * @date: 2016-10-22
17  */
18 public class ReceiverWithTopicExchange {
19     // 交换器名称
20     private static final String EXCHANGE_NAME = "myTopicExchange";
21 
22     public static void main(String[] args) throws Exception {
23         // 连接到rabbitmq服务器
24         ConnectionFactory factory = new ConnectionFactory();
25         factory.setHost("localhost");
26         Connection connection = factory.newConnection();
27         // 创建一个信道
28         final Channel channel = connection.createChannel();
29         // 定义一个名字为topicExchange的topic类型的exchange
30         channel.exchangeDeclare(EXCHANGE_NAME, "topic");
31         
32         //定义接收警告消息的队列
33         channel.queueDeclare("warningQueue", false, false, false, null);
34         //定义接收错误消息的队列
35         channel.queueDeclare("errorQueue", false, false, false, null);
36         //定义接收所有级别日志消息的队列
37         channel.queueDeclare("allLoggerQueue", false, false, false, null);
38         
39         //使用logger.warning路由键绑定myTopicExchange与warningQueue
40         channel.queueBind("warningQueue", EXCHANGE_NAME, "logger.warning");
41         //使用logger.error路由键绑定myTopicExchange与errorQueue
42         channel.queueBind("errorQueue", EXCHANGE_NAME, "logger.error");
43         //使用logger.*路由规则绑定myTopicExchange与allLoggerQueue
44         channel.queueBind("allLoggerQueue", EXCHANGE_NAME, "logger.*");
45         
46         //只处理警告日志，使用手动ack确认
47         channel.basicConsume("warningQueue", false, new DefaultConsumer(channel){
48             @Override
49             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
50                     throws IOException {
51                 String msg = new String(body);
52                 System.out.println("warningQueue accept a warning msg :" + msg);
53                 channel.basicAck(envelope.getDeliveryTag(), false);
54             }
55         });
56         //只处理错误日志，使用手动ack确认
57         channel.basicConsume("errorQueue", false, new DefaultConsumer(channel){
58             @Override
59             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
60                     throws IOException {
61                 String msg = new String(body);
62                 System.out.println("errorQueue accept a error msg :" + msg);
63                 channel.basicAck(envelope.getDeliveryTag(), false);
64             }
65         });
66         //处理警告和错误日志，使用手动ack确认
67         channel.basicConsume("allLoggerQueue", false, new DefaultConsumer(channel){
68             @Override
69             public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
70                     throws IOException {
71                 String msg = new String(body);
72                 System.out.println("allLoggerQueue accept a logger msg :" + msg);
73                 channel.basicAck(envelope.getDeliveryTag(), false);
74             }
75         });
76     }
77 }

发送警告消息：warning message is :1
发送错误消息：error message is :1
发送警告消息：warning message is :2
发送错误消息：error message is :2
发送错误消息：error message is :3
发送警告消息：warning message is :3

allLoggerQueue accept a logger msg :error message is :1
allLoggerQueue accept a logger msg :warning message is :1
errorQueue accept a error msg :error message is :1
warningQueue accept a warning msg :warning message is :1
warningQueue accept a warning msg :warning message is :2
errorQueue accept a error msg :error message is :2
allLoggerQueue accept a logger msg :warning message is :2
allLoggerQueue accept a logger msg :error message is :2
allLoggerQueue accept a logger msg :warning message is :3
errorQueue accept a error msg :error message is :3
warningQueue accept a warning msg :warning message is :3
allLoggerQueue accept a logger msg :error message is :3