您好,登錄后才能下訂單哦!
在我們的日志系統中,我們不只希望訂閱基于嚴重程度的日志,同時還希望訂閱基于發送來源的日志。Unix 工具 syslog 就是同時基于嚴重程度 -severity (info/warn/crit…) 和 設備 -facility (auth/cron/kern…) 來路由日志的。
如果這樣的話,將會給予我們非常大的靈活性,我們既可以監聽來源于”cron” 的嚴重程度為”critical errors” 的日志,也可以監聽來源于”kern” 的所有日志。
為了實現這個目的,接下來我們學習如何使用另一種更復雜的交換器 ——topic exchange。
Topic exchange
topic exchange 與 direct exchange 類似,也是將消息路由到 binding key 與 routing key 相匹配的 Queue 中,但這里的匹配規則有些不同,它約定:
routing key 為一個句點號. 分隔的字符串(我們將被句點號. 分隔開的每一段獨立的字符串稱為一個單詞),如 “stock.usd.nyse”、”nyse.vmw”、”quick.orange.rabbit”
binding key 與 routing key 一樣也是句點號. 分隔的字符串
binding key 中可以存在兩種特殊字符 * 與#,用于做模糊匹配,其中 * 用于匹配一個單詞,#用于匹配多個單詞(可以是零個)
這個例子里,我們發送的所有消息都是用來描述小動物的。發送的消息所攜帶的路由鍵是由三個單詞所組成的,這三個單詞被兩個。分割開。路由鍵里的第一個單詞描述的是動物的速度,第二個單詞是動物的顏色,第三個是動物的種類。所以它看起來是這樣的:” 速度。顏色。種類”。
我們創建了三個綁定:Q1 的 binding key 為”.orange.“,Q2 的 binding key 為”…rabbit” 和”lazy.#”。
這三個 binding key 被可以總結為:
Q1 對所有的桔黃色動物都感興趣。
Q2 則是對所有的兔子和所有懶惰的動物感興趣。
以上圖中的配置為例:
routingKey=”quick.orange.rabbit” 的消息會同時路由到 Q1 與 Q2
routingKey=”lazy.orange.fox” 的消息會路由到 Q1 與 Q2
routingKey=”lazy.brown.fox” 的消息會路由到 Q2
routingKey=”lazy.pink.rabbit” 的消息會路由到 Q2(只會投遞給 Q2 一次,雖然這個 routingKey 與 Q2 的兩個 bindingKey 都匹配)
routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit” 的消息將會被丟棄,因為它們沒有匹配任何 bindingKey
如果我們違反約定,發送了一個 routing key 為一個單詞或者四個單詞(”orange” or “quick.orange.male.rabbit”)的消息時,該消息不會投遞給任何一個隊列,而且會丟失掉。
了解springcloud架構可以加求求:三五三六二四七二五九
但是,即使”lazy.orange.male.rabbit” 有四個單詞,他還是會匹配最后一個綁定,并且被投遞到第二個隊列中。
topic exchange
topic exchange 是強大的,它可以表現出跟其他 exchange 類似的行為。
當一個隊列的 binding key 為 “#”(井號) 的時候,它會接收所有消息,而不考慮 routing key,就像 fanout exchange。
當 * (星號) 和 # (井號) 這兩個特殊字符都未在綁定鍵中出現的時候,此時 topic exchange 會表現得像 direct exchange 一樣。
代碼整合
生產者
public class Tut5Sender { @Autowird private AmqpTemplate template; @Autowird private TopicExchange topic; private int index; private int count; private final String[] keys = {"quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox", "lazy.brown.fox", "lazy.pink.rabbit", "quick.brown.fox"}; @Scheduled(fixedDelay = 1000, initialDelay = 500) public void send() { StringBuilder builder = new StringBuilder("Hello to "); if (++this.index == keys.length) { this.index = 0; } String key = keys[this.index]; builder.append(key).append(' '); builder.append(Integer.toString(++this.count)); String message = builder.toString(); template.convertAndSend(topic.getName(), key, message); System.out.println(" [x] Sent '" + message + "'"); } }
消費者
public class Tut5Receiver { @RabbitListener(queues = "#{autoDeleteQueue1}") public void receiver1(String in) throws InterruptedException { receive(in, 1); } @RabbitListener(queues = "#{autoDeleteQueue2}") public void receiver2(String in) throws InterruptedException { receive(in, 2); } private void receive(String in, int instance) throws InterruptedException { StopWatch watch = new StopWatch(); watch.start(); System.out.println("instance " + instance + " [x] Received '" + in + "'"); doWork(in); watch.stop(); System.out.println("instance " + instance + " [x] Done in " + watch.getTotalTimeSeconds() + "s"); } private void doWork(String in) throws InterruptedException { for (char c : in.toCharArray()) { if (c == '.') { Thread.sleep(1000); } } } }
配置類
@Profile({"tut5", "topics"}) @Configuration public class Tut5Config { @Bean public TopicExchange topic() { return new TopicExchange("tut.topic"); } @Profile("receiver") private static class ReceiverConfig { @Bean public Queue autoDeleteQueue1() { return new AnonymousQueue(); } @Bean public Queue autoDeleteQueue2() { return new AnonymousQueue(); } @Bean public Binding binding1a(TopicExchange topic, Queue autoDeleteQueue1) { return BindingBuilder.bind(autoDeleteQueue1) .to(topic) .with("*.orange.*"); } @Bean public Binding binding2a(TopicExchange topic, Queue autoDeleteQueue2) { return BindingBuilder.bind(autoDeleteQueue2) .to(topic) .with("*.*.rabbit"); } @Bean public Binding binding2b(TopicExchange topic, Queue autoDeleteQueue2) { return BindingBuilder.bind(autoDeleteQueue2) .to(topic) .with("lazy.#"); } @Bean public Tut5Receiver receiver() { return new Tut5Receiver(); } } @Profile("sender") @Bean public Tut5Sender sender() { return new Tut5Sender(); } }
運行
maven 編譯
mvn clean package -Dmaven.test.skip=true
運行
java -jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar --spring.profiles.active=tut5,receiver --tutorial.client.duration=60000
java -jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar --spring.profiles.active=tut5,sender --tutorial.client.duration=60000
輸出
// Sender
Ready … running for 60000ms
[x] Sent ‘Hello to lazy.orange.elephant 1’
[x] Sent ‘Hello to quick.orange.fox 2’
[x] Sent ‘Hello to lazy.brown.fox 3’
[x] Sent ‘Hello to lazy.pink.rabbit 4’
[x] Sent ‘Hello to quick.brown.fox 5’
[x] Sent ‘Hello to quick.orange.rabbit 6’
[x] Sent ‘Hello to lazy.orange.elephant 7’
[x] Sent ‘Hello to quick.orange.fox 8’
[x] Sent ‘Hello to lazy.brown.fox 9’
[x] Sent ‘Hello to lazy.pink.rabbit 10’
// Receiver
Ready … running for 60000ms
instance 1 [x] Received ‘Hello to lazy.orange.elephant 1’
instance 2 [x] Received ‘Hello to lazy.orange.elephant 1’
instance 2 [x] Done in 2.004s
instance 1 [x] Done in 2.004s
instance 2 [x] Received ‘Hello to lazy.brown.fox 3’
instance 1 [x] Received ‘Hello to quick.orange.fox 2’
instance 1 [x] Done in 2.006s
instance 2 [x] Done in 2.006s
instance 2 [x] Received ‘Hello to lazy.pink.rabbit 4’
instance 1 [x] Received ‘Hello to quick.orange.rabbit 6’
instance 2 [x] Done in 2.006s
instance 2 [x] Received ‘Hello to quick.orange.rabbit 6’
instance 1 [x] Done in 2.007s
instance 1 [x] Received ‘Hello to lazy.orange.elephant 7’
instance 2 [x] Done in 2.006s
instance 2 [x] Received ‘Hello to lazy.orange.elephant 7’
instance 1 [x] Done in 2.003s
instance 1 [x] Received ‘Hello to quick.orange.fox 8’
instance 2 [x] Done in 2.005s
instance 2 [x] Received ‘Hello to lazy.brown.fox 9’
instance 1 [x] Done in 2.005s
instance 2 [x] Done in 2.004s
instance 2 [x] Received ‘Hello to lazy.pink.rabbit 10’
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。