spring boot使用RabbitMQ实现topic 主题

时间:2022-12-07 10:01:03

前一篇我们实现了消息系统的灵活配置。代替了使用扇形(fanout)交换器的配置。使用直连(direct)交换器,并且基于路由键后可以有选择性接收消息的能力。

虽然使用直连交换器可以改善我们的系统,但是它仍有局限性,它不能实现多重条件的路由。

在我们的消息系统中,我们不仅想要订阅基于路由键的队列,还想订阅基于生产消息的源。这些概念来自于unix工具syslog。该日志基于严格的(info/warn/crit...) 和容易的(auth/cron/kern...)的路由方式。我们的例子比这个要简单。

这个例子将会给我们很大的灵活性,比如我们既想监听来‘cron'自错误的日志又想监听来自‘kern'的所有日志。

为了实现这个灵活性,我们需要知道更多关于主题交换器的内容。

主题交换器

使用主题交换器时不能采用任意写法的路由键,路由键的形式应该是由点分割的单词。用什么词都行,通常都是能表明意义的。例如"stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit"。但字数大小被限制在最多255字节。

使用主题交换器定义路由键需要注意点2点

  1. *星号代表一个单词。
  2. #井号代表0个或多个单词。

spring boot使用RabbitMQ实现topic 主题

定义符合主题交换器的路由键

在这个例子中,我们将发送所有描述动物的消息。这个消息将会和由3个单词2个点构成的路由键一起发送。第一个单词是表述速度,第二个描述颜色,第三个描述种类:"<speed>.<colour>.<species>"。

创建三种绑定,q1和键"*.orange.*"绑定,q2和"*.*.rabbit" 、"lazy.#"绑定。

三种绑定关系的概述为:

  1. q1 对橙色的动物感兴趣。(队列1)
  2. q2 对所有关于兔子和所有关于慢速的动物感兴趣。(队列2)

一个和路由键被设置成"quick.orange.rabbit"的消息将会被传递到q1、q2这两个队列中。"lazy.orange.elephant" 也会这样。"quick.orange.fox"会去第一个队列,"lazy.brown.fox"会去第二个队列,"lazy.pink.rabbit"会去第二个队列及时它匹配了2次绑定。"quick.brown.fox"因为不匹配哪也去不了,会被丢弃。

那么像"orange" 、 "quick.orange.male.rabbit"这样的呢?因为没有匹配到任何绑定也会被丢弃。

那么像"lazy.orange.male.rabbit"也是四个词的路由键呢?,由于匹配到了lazy.#这个将会被传递到第二个队列中。

主题交换器的小技巧

主题交换器是牛逼的并且表现的与其它交换器相似。

  1. 当一个队列和 "#" 绑定键绑定时,该队列能收到所有的消息,这点与扇形(fanout)交换器类似。
  2. 当不使用 "*" and "#" 时,主题交换器就与直连交换器没啥两样。

代码示例

代码与之前的路由代码没啥两样,请看

config.java

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
package com.zb.rabbitmqtest.t5topics.config;
 
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.bean;
import org.springframework.context.annotation.configuration;
 
/**
 * @author 张博
 */
@configuration(value = "t5config")
public class config {
 
  /**
   * 创建人:张博
   * 时间:2018/3/5 上午10:45
   * @apinote 定义主题交换器
   */
  @bean
  public topicexchange topicexchange() {
    return new topicexchange("topic-exchange");
  }
 
  /**
   * 创建人:张博
   * 时间:2018/3/5 上午10:48
   * @apinote 定义自动删除匿名队列
   */
  @bean
  public queue autodeletequeue0() {
    return new anonymousqueue();
  }
 
  /**
   * 创建人:张博
   * 时间:2018/3/5 上午10:48
   * @apinote 定义自动删除匿名队列
   */
  @bean
  public queue autodeletequeue1() {
    return new anonymousqueue();
  }
 
  /**
   * 创建人:张博
   * 时间:2018/3/5 上午10:48
   * @param topicexchange 主题交换器
   * @param autodeletequeue0 自动删除队列
   * @apinote 绑定使用路由键为 orange 的 autodeletequeue0 队列到主题交换器上
   * @return binding
   */
  @bean
  public binding binding0a(topicexchange topicexchange, queue autodeletequeue0) {
    return bindingbuilder.bind(autodeletequeue0).to(topicexchange).with("*.orange.*");
  }
 
  /**
   * 创建人:张博
   * 时间:2018/3/5 上午10:48
   * @param topicexchange 主题交换器
   * @param autodeletequeue1 自动删除队列
   * @apinote 绑定使用路由键为 black 的 autodeletequeue1 队列到主题交换器上
   * @return binding
   */
  @bean
  public binding binding1a(topicexchange topicexchange, queue autodeletequeue1) {
    return bindingbuilder.bind(autodeletequeue1).to(topicexchange).with("*.*.rabbit");
  }
 
  /**
   * 创建人:张博
   * 时间:2018/3/5 上午10:48
   * @param topicexchange 主题交换器
   * @param autodeletequeue1 自动删除队列
   * @apinote 绑定使用路由键为 green 的 autodeletequeue1 队列到主题交换器上
   * @return binding
   */
  @bean
  public binding binding1b(topicexchange topicexchange, queue autodeletequeue1) {
    return bindingbuilder.bind(autodeletequeue1).to(topicexchange).with("lazy.#");
  }
}

receiver.java

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
package com.zb.rabbitmqtest.t5topics.receiver;
 
import org.springframework.amqp.rabbit.annotation.rabbitlistener;
import org.springframework.stereotype.component;
 
/**
 * @author 张博
 */
@component(value = "t5receiver")
public class receiver {
 
  @rabbitlistener(queues = "#{autodeletequeue0.name}")
  public void receiver0(string str) {
    system.out.println("receiver0++++++++++:" + str);
    //try {
    //  thread.sleep(1000);
    //} catch (interruptedexception e) {
    //  e.printstacktrace();
    //}
  }
 
  @rabbitlistener(queues = "#{autodeletequeue1.name}")
  public void receiver1(string str) {
    system.out.println("receiver1++++++++++:" + str);
    //try {
    //  thread.sleep(1000);
    //} catch (interruptedexception e) {
    //  e.printstacktrace();
    //}
  }
}

send.java

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
package com.zb.rabbitmqtest.t5topics.send;
 
import org.springframework.amqp.core.topicexchange;
import org.springframework.amqp.rabbit.core.rabbittemplate;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.component;
 
/**
 * @author 张博
 */
@component(value = "t5send")
public class send {
 
  @autowired
  private topicexchange topicexchange;
 
  @autowired
  private rabbittemplate rabbittemplate;
 
  private string[] keys = {"quick.orange.rabbit",
      "lazy.orange.elephant", "quick.orange.fox",
      "lazy.brown.fox", "lazy.pink.rabbit", "quick.brown.fox"};
 
  public void send() {
    string message = "哈哈哈";
    for (int i = 0; i < 5; i++) {
      system.out.println("send++++++++++:".concat(message));
      rabbittemplate.convertandsend(topicexchange.getname(), keys[5], message);
    }
  }
}

sendtest.java

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
package com.zb.rabbitmqtest.t5topics.send;
 
import org.junit.test;
import org.junit.runner.runwith;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.boot.test.context.springboottest;
import org.springframework.test.context.junit4.springrunner;
 
/**
 * @author 张博
 */
@runwith(springrunner.class)
@springboottest
public class sendtest {
 
  @autowired
  private send send;
  @test
  public void send() throws exception {
    send.send();
  }
}

测试结果我就不放了 大家请自行查看。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持服务器之家。

原文链接:https://www.jianshu.com/p/66c0072e9bf4