(2)关于RabbitMq 的 Topic Exchange 主题交换机
Topic Exchange 主题交换机
- 一、项目里面创建TopicRabbitConfig
- 二、然后添加多2个接口,用于推送消息到主题交换机:
- 三、创建TopicManReceiver
- 总结
一、项目里面创建TopicRabbitConfig
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 TopicRabbitConfig {
//绑定键
public final static String man = "topic.man";
public final static String woman = "topic.woman";
@Bean
public Queue firstQueue() {
return new Queue(TopicRabbitConfig.man);
}
@Bean
public Queue secondQueue() {
return new Queue(TopicRabbitConfig.woman);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("topicExchange");
}
//将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
//这样只要是消息携带的路由键是topic.man,才会分发到该队列
@Bean
Binding bindingExchangeMessage() {
return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
}
//将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
@Bean
Binding bindingExchangeMessage2() {
return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
}
}
二、然后添加多2个接口,用于推送消息到主题交换机:
@GetMapping("/sendTopicMessage1")
public String sendTopicMessage1() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: M A N ";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> manMap = new HashMap<>();
manMap.put("messageId", messageId);
manMap.put("messageData", messageData);
manMap.put("createTime", createTime);
rabbitTemplate.convertAndSend("topicExchange", "topic.man", manMap);
return "ok";
}
@GetMapping("/sendTopicMessage2")
public String sendTopicMessage2() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: woman is all ";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> womanMap = new HashMap<>();
womanMap.put("messageId", messageId);
womanMap.put("messageData", messageData);
womanMap.put("createTime", createTime);
rabbitTemplate.convertAndSend("topicExchange", "topic.woman", womanMap);
return "ok";
}
}
三、创建TopicManReceiver
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Map;
@Component
public class TopicManReceiver {
@RabbitHandler
@RabbitListener(queues = "topic.man")
public void process(Map testMessage) {
System.out.println("TopicManReceiver消费者收到消息 : " + testMessage.toString());
}
@RabbitHandler
@RabbitListener(queues = "topic.woman")
public void process(Map testMessage) {
System.out.println("TopicManReceiver消费者收到消息 : " + testMessage.toString());
}
}
总结
:先调用/sendTopicMessage1
然后看消费者rabbitmq-consumer的控制台输出情况:
TopicManReceiver监听队列1,绑定键为:topic.man
TopicTotalReceiver监听队列2,绑定键为:topic.#
而当前推送的消息,携带的路由键为:topic.man
所以可以看到两个监听消费者receiver都成功消费到了消息,因为这两个recevier监听的队列的绑定键都能与这条消息携带的路由键匹配上。
接下来调用接口/sendTopicMessage2:
然后看消费者rabbitmq-consumer的控制台输出情况:
TopicManReceiver监听队列1,绑定键为:topic.man
TopicTotalReceiver监听队列2,绑定键为:topic.#
而当前推送的消息,携带的路由键为:topic.woman
所以可以看到两个监听消费者只有TopicTotalReceiver成功消费到了消息。
到此Topic Exchange 主题交换机就算是完事了
下篇我们使用Fanout Exchang 扇型交换机