Topic交换器
1、示例
交换器：交换器topic详情：临时消息队列：发送消息：接收消息：
Topic交换器 发送到topic交换器的消息不能有任意的routing_key，它必须是单词列表，由点分隔。
示例： “stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
路由键中可以由任意多的单词，最多为255个字节
topic交换器背后的逻辑类似direct交换器——使用特定路由键发送的消息将被传递到与匹配绑定键绑定的所有队列。但是，绑定键有两个重要的特殊情况：
*（星号）可以只替换一个单词。# (hash) 可以代替零个或多个单词。
在这个例子中，我们将发送所有描述动物的消息。消息将使用由三个单词（两个点）组成的路由键发送。路由键中的第一个词将描述速度，第二个是颜色，第三个是物种：“ .. ”。
我们创建了三个绑定：Q1 与绑定键“ .orange. ”绑定，Q2 与“ ..rabbit ”和“ lazy.# ”绑定。
这些绑定可以概括为：
Q1 对所有橙色动物都感兴趣。Q2 想听听关于兔子的一切，以及关于懒惰动物的一切。
路由键设置为“ 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 ”，即使它有四个单词，也会匹配最后一个绑定，并被传递到第二个队列。
1、示例 交换器： 交换器topic详情： 临时消息队列： 发送消息：主题交换功能强大，可以像其他交换一样运行。
当队列与“ # ”（散列）绑定键绑定时 - 无论路由键如何，它将接收所有消息 - 就像在fanout交换中一样。当绑定中不使用特殊字符“ * ”（星号）和“ # ”（哈希）时，topic交换的行为就像direct交换一样。
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) throws Exception {
// 定义参数
argv = new String[]{"*.critical","hello,topicExchange"};
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.6.128");
factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("admin");
try (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("UTF-8"));
System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
}
}
private static String getRouting(String[] strings) {
if (strings.length < 1)
return "anonymous.info";
return strings[0];
}
private static String getMessage(String[] strings) {
if (strings.length < 2)
return "Hello World!";
return joinStrings(strings, " ", 1);
}
private static String joinStrings(String[] strings, String delimiter, int startIndex) {
int length = strings.length;
if (length == 0) return "";
if (length < startIndex) return "";
StringBuilder words = new StringBuilder(strings[startIndex]);
for (int i = startIndex + 1; i < length; i++) {
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
接收消息：
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) throws Exception {
// 定义参数
argv = new String[]{"*.critical","critical.#"};
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.6.128");
factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("admin");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 定义交换器名称及类型
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
// 获取一个临时的队列
String queueName = channel.queueDeclare().getQueue();
if (argv.length < 1) {
System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
System.exit(1);
}
for (String bindingKey : argv) {
// 将交换器、队列、路由键名绑定
channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
}
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
