003 topic

TopicRabbitmqConfig.java

@Bean：这是Spring的一个注解，用于告诉Spring这个方法会返回一个对象，该对象应被注册为Spring应用上下文中的一个Bean。
public TopicExchange newTopicExchange()：定义了一个公开的方法newTopicExchange，该方法返回一个TopicExchange对象。
new TopicExchange(“topicExchange”,true,false)：创建一个名为"topicExchange"的Topic交换机。第二个参数true表示交换机是否持久化，第三个参数false表示交换机是否自动删除（当没有队列与其绑定时）。

定义一个名为bindingQueue011的方法，该方法返回一个Binding对象。
BindingBuilder.bind(newQueue011()).to(newTopicExchange()).with(“aaa.ccc”)：创建一个绑定，将"topicQueue011"队列绑定到"topicExchange"交换机，并设置路由键为"aaa.ccc"。

定义一个名为bindingQueue022的方法，该方法返回一个Binding对象。
BindingBuilder.bind(newQueue022()).to(newTopicExchange()).with(“aaa.")：创建一个绑定，将"topicQueue022"队列绑定到"topicExchange"交换机，并设置路由键为"aaa.”。这里的路由键使用了通配符*，表示任意一个单词，这意味着发送到交换机的消息，只要其路由键是以"aaa."开头并后跟任意一个单词，都会被路由到这个队列。

package com.example.rabbitmq.topic;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TopicRabbitmqConfig {
    @Bean
    public TopicExchange newTopicExchange(){
        return new TopicExchange("topicExchange",true,false);
    }

    @Bean
    public Queue newQueue011(){
        return new Queue("topicQueue011",true);
    }
    @Bean
    public Queue newQueue022(){
        return new Queue("topicQueue022",true);

    }

    @Bean
    public Binding bindingQueue011(){
        return BindingBuilder.bind(newQueue011()).to(newTopicExchange()).with("aaa.ccc");
    }
    @Bean
    public Binding bindingQueue022(){
        return BindingBuilder.bind(newQueue022()).to(newTopicExchange()).with("aaa.*");

        // aaa.# : # 可以是任意一个或多个单词 ， aaa.bbb, aaa.bbb.ccc,aaa
        // aaa.* : * 只能是任意一个单词  aaa.bbb（ok）,aaa.bbb.ccc(no)

    }
}



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

TopicProducer.java

总的来说，这个TopicProducer类是一个Spring REST控制器，它提供了一个HTTP端点/topic/sendMsg，当调用这个端点时，会向RabbitMQ的topicExchange交换机发送两条带有不同路由键的消息。

rabbitTemplate.convertAndSend(…)：这是RabbitTemplate的一个方法，用于将消息发送到RabbitMQ。
第一个参数"topicExchange"：指定要发送到的交换机名称。
第二个参数"aaa.bbb"：是消息的路由键（Routing Key），它决定了消息应该如何被路由到不同的队列。
第三个参数"消息1"：是要发送的实际消息内容。

package com.example.rabbitmq.topic;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("topic")
public class TopicProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("sendMsg")
    public String  sendMsg(){
        String msg ="hello topic msg111";
        rabbitTemplate.convertAndSend("topicExchange","aaa.bbb","消息1");  //q2
        rabbitTemplate.convertAndSend("topicExchange","aaa.bbb.cccc","消息2");
        return "ok";
    }

}



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

TopicConsumer.java

@RabbitListener：这是Spring AMQP提供的一个注解，用于监听指定的队列。
queues = “topicQueue011”：这个参数指定了要监听的队列名称，即topicQueue011。

public void processQueue011(String msg){
定义了一个公开的方法processQueue011，该方法接收一个字符串参数msg，这个参数代表从topicQueue011队列中接收到的消息。

总的来说，TopicConsumer类是一个Spring组件，它定义了两个方法来分别处理来自topicQueue011和topicQueue022两个队列的消息。当RabbitMQ中有消息到达这两个队列时，相应的方法会被调用，并打印出消息内容。

package com.example.rabbitmq.topic;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class TopicConsumer {
    @RabbitHandler
    @RabbitListener(queues = "topicQueue011")
    public  void processQueue011(String msg){
        System.out.println("消费者接收到的队列011里面的消息：" + msg);
    }

    @RabbitHandler
    @RabbitListener(queues = "topicQueue022")
    public  void processQueue022(String msg){
        System.out.println("消费者接收到的队列022里面的消息：" + msg);
    }
}



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24