之前我们说到吧 exchange 的类型从 fanout 换到了 direct, 然后就可以选择性的接收并发送 message 了. 尽管 direct 可以满足我们的需求提高了当前的系统可用性,但是他还有局限性. 它并不能建立在多个路由规则上. 而实际上我们有可能面对多个 log 的源, 比方说 unix 系统上来自于系统本身的,还有 cron 等 log 什么的,这个情况下就相对追加了复杂性. 这里我们拿kernel 和cron 来举例.

而这里为了能够实现这个需求我们就需要理解并学习topic 类型的 exchange 了.

topic exchange

当一个 exchange 他是 topic 类型的时候, message 本身就不能够简单的拿routing_key 来完成了. 它必须是一些有 dot 拼接起来的字符串来表示, 这些字符可以是任意内容. 但是通常他们都是和当前的消息内容本身有关系,能够表达这个消息是什么内容. 比如说"stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit". 只要不超过255字节长度都可以的.

绑定键同样和路由键是一样的格式,对于 topic 类型的 exchange 其背后的逻辑同样和 direct 类型的一样, 也就是一个消息会通过特定的路由键被发送到 所有已经通过特定的绑定键绑定好的queue 上. 但是也有两个比较特定的规定就是:

* 号可以替代任意一个单词
# hash 号可以替代任意多个单词

在这个例子里, 我们将要面对所有描述动物的消息,消息的路由键是有三个单词(2个 dot) 组成的.一个单词会描述速度, 第二个是颜色,第三个是物种. "..".

我们创建3个绑定. ".orange." and Q2 with "..rabbit" and "lazy.#".

这些绑定可以归纳为:

Q1是对于 orange 动作感兴趣
Q2对于 rabbit 的所有感兴趣. 还有就是 lazy 的动物

这里, 如果一个 message 它的路由 key 是"quick.orange.rabbit"那么它将会被发往2个 queue. 还有如 "lazy.orange.elephant" 也是会发往2个 queue.
换句话说 On the other hand "quick.orange.fox" will only go to the first queue, and "lazy.brown.fox" only to the second.

"lazy.pink.rabbit" will be delivered to the second queue only once, even though it matches two bindings. "quick.brown.fox" doesn't match any binding so it will be discarded.

like "orange" or "quick.orange.male.rabbit"? Well, these messages won't match any bindings and will be lost.

On the other hand "lazy.orange.male.rabbit", even though it has four words, will match the last binding and will be delivered to the second queue.

关于 Topic exchange

topic exchange 其实非常强大, 它可以拿来以其他类型的 exchange 使用, 如果当一个 queue 通过 #来绑定,那么它将会接收到所有的消息, 和路由键就无关了. 这个场合就和 fanout 是一样的.
而如果在绑定键里并没有使用※或者hash 号那就盒 direct 是一样的.

#!/usr/bin/env ruby
# encoding: utf-8

require "bunny"

conn = Bunny.new # 建立连接
conn.start

ch       = conn.create_channel # 创建 channel
x        = ch.topic("topic_logs") #  exchange
severity = ARGV.shift || "anonymous.info"
msg      = ARGV.empty? ? "Hello World!" : ARGV.join(" ")

x.publish(msg, :routing_key => severity) # 拿路由 key来发送消息
puts " [x] Sent #{severity}:#{msg}"

conn.close

############The code for receive_logs_topic.rb:

#!/usr/bin/env ruby
# encoding: utf-8

require "bunny"

if ARGV.empty?
  abort "Usage: #{$0} [binding key]"
end

conn = Bunny.new
conn.start

ch  = conn.create_channel
x   = ch.topic("topic_logs")
q   = ch.queue("", :exclusive => true) # 创建临时 queue

ARGV.each do |severity|
  q.bind(x, :routing_key => severity) # 通过路由键绑定到 exchange
end

puts " [*] Waiting for logs. To exit press CTRL+C"

begin
  q.subscribe(:block => true) do |delivery_info, properties, body|
    puts " [x] #{delivery_info.routing_key}:#{body}"
  end
rescue Interrupt => _
  ch.close
  conn.close
end

others:

ref:

再说 rabbitmq 续4

topic exchange

关于 Topic exchange