topic的路由规则里使用【.】号分隔单词,使用【*】号匹配1个单词,使用【#】匹配多个.和多个*。

在下面的例子中:

logger.*可以匹配logger.error和logger.warning,但logger*.error只能匹配logger.error

logger#可以匹配到logger.error和logger.warning。

下面的例子使用topic接收警告、错误的日志,并根据匹配的路由规则发送给不同的Queue队列来处理的例子:

日志生产者SenderWithTopicExchange

 package com.yzl.test2;

 import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory; /**
* 使用topic交换器发送消息
* 分为警告和错误2种日志
* @author: yzl
* @date: 2016-10-22
*/
public class SenderWithTopicExchange {
//交换器名称
private static final String EXCHANGE_NAME = "myTopicExchange";
//路由键的前缀
private static final String BASE_ROUTING_KEY = "logger."; public static void main(String[] args) throws Exception {
//使用CountDownLatch控制2个线程一起运行
final CountDownLatch cdl = new CountDownLatch(2);
//连接到rabbitmq服务器
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
//创建一个信道
final Channel channel = connection.createChannel();
//定义一个名字为topicExchange的topic类型的exchange
channel.exchangeDeclare(EXCHANGE_NAME, "topic"); ExecutorService pool = Executors.newFixedThreadPool(2);
pool.submit(new Runnable() {
@Override
public void run() {
try {
cdl.await();
//发送警告日志,绑定路由键:logger.warning
String warningMsg = "warning message is :";
for(int i=1; i<800; i++){
System.out.println("发送警告消息:" + warningMsg+i);
channel.basicPublish(EXCHANGE_NAME, BASE_ROUTING_KEY + "warning", null, (warningMsg+i).getBytes());
Thread.sleep(2000L);
}
} catch (Exception e) {
e.printStackTrace();
}
}
});
pool.submit(new Runnable() {
@Override
public void run() {
try {
cdl.await();
//发送错误日志,绑定路由键:logger.error
String errorMsg = "error message is :";
for(int i=1; i<1000; i++){
System.out.println("发送错误消息:" + errorMsg+i);
channel.basicPublish(EXCHANGE_NAME, BASE_ROUTING_KEY + "error", null, (errorMsg+i).getBytes());
Thread.sleep(2000L);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}); cdl.countDown();
cdl.countDown();
}
}

消息消费者ReceiverWithTopicExchange

 package com.yzl.test2;

 import java.io.IOException;

 import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope; /**
* 使用topic交换器接收消息
*
* @author: yzl
* @date: 2016-10-22
*/
public class ReceiverWithTopicExchange {
// 交换器名称
private static final String EXCHANGE_NAME = "myTopicExchange"; public static void main(String[] args) throws Exception {
// 连接到rabbitmq服务器
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
// 创建一个信道
final Channel channel = connection.createChannel();
// 定义一个名字为topicExchange的topic类型的exchange
channel.exchangeDeclare(EXCHANGE_NAME, "topic"); //定义接收警告消息的队列
channel.queueDeclare("warningQueue", false, false, false, null);
//定义接收错误消息的队列
channel.queueDeclare("errorQueue", false, false, false, null);
//定义接收所有级别日志消息的队列
channel.queueDeclare("allLoggerQueue", false, false, false, null); //使用logger.warning路由键绑定myTopicExchange与warningQueue
channel.queueBind("warningQueue", EXCHANGE_NAME, "logger.warning");
//使用logger.error路由键绑定myTopicExchange与errorQueue
channel.queueBind("errorQueue", EXCHANGE_NAME, "logger.error");
//使用logger.*路由规则绑定myTopicExchange与allLoggerQueue
channel.queueBind("allLoggerQueue", EXCHANGE_NAME, "logger.*"); //只处理警告日志,使用手动ack确认
channel.basicConsume("warningQueue", false, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
String msg = new String(body);
System.out.println("warningQueue accept a warning msg :" + msg);
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
//只处理错误日志,使用手动ack确认
channel.basicConsume("errorQueue", false, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
String msg = new String(body);
System.out.println("errorQueue accept a error msg :" + msg);
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
//处理警告和错误日志,使用手动ack确认
channel.basicConsume("allLoggerQueue", false, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
String msg = new String(body);
System.out.println("allLoggerQueue accept a logger msg :" + msg);
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}

结果输出:

发送警告消息: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

消息处理流程:

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

04-26 18:37
查看更多