rabbitMQ_topic(五)
主题转发器
发送到主题转发器的消息不能有任意的 routing_key - 它必须是由点分隔的单词列表。这些单词可以是任何东西,但通常它们指定与消息相关联的一些功能。几个有效的routeKey示例:“ stock.usd.nyse”,“ nyse.vmw ”,“ quick.orange.rabbit ”。routeKey中可以有任意多的单词,最多可达255个字节。
绑定键(即绑定时使用的routeKey)也必须是相同的形式。topics转发器可以是用通配符:
- *(星)可以替代一个单词。
- #(哈希)可以替换零个或多个单词。
如下图:
在这个例子中,我们将发送所有描述动物的消息。消息将使用由三个字(两个点)组成的routeKey发送。routeKey中的第一个单词将描述速度,第二个描述颜色,第三个描述种类( <speed>.<color>.<species>)。
我们创建了三个绑定:Q1队列绑定键* .orange.*,Q2绑定键*.*.rabbit和lazy.#。
这些绑定可以总结为:
- Q1对所有的橙色动物感兴趣。
- Q2想听听有关兔子的一切,以及关于懒惰动物的一切。
将routeKey设置为“ quick.orange.rabbit ”的消息将传递给两个队列。消息“ lazy.orange.elephant ”也会传递给两个队列。“ quick.orange.fox ”只会转到第一个队列,而“ lazy.brown.fox ”只能到第二个队列。“ lazy.pink.rabbit ”将被传递到第二个队列。“ quick.brown.fox ”不匹配任何绑定,因此它将被丢弃。
如果我们不按套路出牌,发送一个或四个字的消息,如“ orange ”或“ quick.orange.male.rabbit ”,会发生什么?那么这些消息将不会匹配任何绑定,并将丢失。
但是在某种情况下,如“ lazy.orange.male.rabbit ”即使它有四个单词,却会匹配最后的绑定,并将消息传递到第二个队列。
主题转发器
主题转发是强大的,其他转发器能做到的事情,它也能做到。
当队列与“ # ”(哈希)绑定键绑定时,它将接收所有消息,而不管绑定键是什么,就像fanout转发器一样。
当特殊字符“ * ”(星号)和“ # ”(哈希)在绑定中不被使用时,主题转发器将表现得像一个直接转发器。
完整示例
EmitLogTopic.java的代码:
package com.rabbitMQ; import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory; public class EmitLogTopic { 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 routingKey = getRouting(argv);
String message = getMessage(argv); channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'"); connection.close();
} private static String getRouting(String[] argv) { if (argv.length == ) { return "jr;;p.critical";
}
String str = argv[]; String routeKey = str.replaceAll("=.*", ""); return routeKey;
} private static String getMessage(String[] argv) {
if (argv.length == ) { return "I like play game.";
}
String str = argv[];
String message = str.replaceAll(".*=", "");
return message;
} }
ReceiveLogsTopic.java的代码:
package com.rabbitMQ; import java.io.IOException; import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope; public class ReceiveLogsTopic {
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(); if (argv.length < ) {
System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
System.exit();
} for (String bindingKey : argv) {
channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
} System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
System.out.println(consumerTag);//amq.ctag-yqB24qMf7hEtpyZgR1p4MQ
System.out.println(properties);//#contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
System.out.println(envelope);//Envelope(deliveryTag=1传送标志, redeliver=false重新传送, exchange=topic_logs转发器名称, routingKey=jr;;p.critical)
}
};
channel.basicConsume(queueName, true, consumer);
}
}
启动ReceiveLogsTopic.java,给它传*.error
启动ReceiveLogsTopic.java,给它传*.info
启动EmitLogTopic.java,给它传#=all matches
结果:俩个ReceiveLogsTopic.java会收到all matches
启动ReceiveLogsTopic.java,给它传mistake=error
只有第一个ReceiveLogsTopic.java收到消息
最新文章
- hdu 5901 Count primes
- Moon River
- JS中的constructor与prototype
- 学习java 第1天
- Oracle中的注释
- [GeekBand] 面向对象的设计模式(C++)(2)
- NGINX(六)扩展
- java编程思想笔记(一)——面向对象导论
- ubuntu server 11.10 安装 oracle 10g XE
- 七字真言解读TCP三次握手
- Chipmunk僵尸物理对象的出现和解决(七)
- pip install -r requirements.txt
- sboot mybatis
- BZOJ.1031.[JSOI2007]字符加密(后缀数组)
- 运行用例时,报错Unknow Error:Element xxx is not clickable……的解决方法
- shell脚本中四则运算
- c++11 学习
- 某游戏应用的redis 数据库结构设计(转)
- Django页面重定向
- C# word 类库基本属性介绍