一、前言
Topic模式，又称主题模式，他和路由模式类似，但是比路由模式灵活，绑定在Topic交换机上的队列，可以设置一个带通配符的路由key，比如orange.*，*.black.*等，例如producer发送消息到Topic交换机时，带上路由可以为orange.1,则会将消息发送到orange.*这个队列。需要注意的是，比如队列绑定的路由key是*.black.*，那producer发送消息时，路由一定要black前后都有值才能匹配到，比如1.black.1才行，1.black是不行的。
二、生产者
1.TopicExchangeConfig.java
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 TopicExchangeConfig {
public static final String TOPIC_EXCHANGE = "topic_exchange";
public static final String QUEUE_ORANGE = "queue_orange";
public static final String QUEUE_BLACK = "queue_black";
public static final String ROUTING_KEY_ORANGE = "orange.*";
public static final String ROUTING_KEY_BLACK = "*.black.*";
/**
* 定义一个Topic类型交换机
* @return
*/
@Bean
public TopicExchange topicExchange(){
return new TopicExchange(TOPIC_EXCHANGE);
}
/**
* 创建一个队列'queue_orange'
* @return
*/
@Bean
public Queue queueOrange(){
return new Queue(QUEUE_ORANGE);
}
/**
* 创建一个队列'queue_black'
* @return
*/
@Bean
public Queue queueBlack(){
return new Queue(QUEUE_BLACK);
}
/**
* 将队列'queue_orange'绑定到Topic交换机上，并设置路由key为'orange.*'
* 该队列则可以接收路由key为orange.1，orange.2等样式的消息
* @return
*/
@Bean
public Binding bindingQueueOrange(){
return BindingBuilder.bind(queueOrange()).to(topicExchange()).with(ROUTING_KEY_ORANGE);
}
/**
* 将队列'queue_orange'绑定到Topic交换机上，并设置路由key为'*.black.*'
* 该队列则可以接收路由key为1.black.1，2.black.2等样式的消息
* 注意：发送路由key为1.black是匹配不到这个队列的
* @return
*/
@Bean
public Binding bindingQueueBlack(){
return BindingBuilder.bind(queueBlack()).to(topicExchange()).with(ROUTING_KEY_BLACK);
}
}
2.ProducerController.java
@RestController
@RequestMapping(value = "send")
public class ProducerController {
@Resource
private RabbitTemplate rabbitTemplate;
/**
* 发送路由key为"orange.1"的消息到Topic交换机，交换机会将该消息投递给队列"queue_orange"
*/
@GetMapping("topic/orange1")
public void sendOrange1(){
String msg = "send routing key = orange.1 msg " + new Date().toString();
rabbitTemplate.convertAndSend(TopicExchangeConfig.TOPIC_EXCHANGE, "orange.1", msg);
}
/**
* 发送路由key为"orange.2"的消息到Topic交换机，交换机会将该消息投递给队列"queue_orange"
*/
@GetMapping("topic/orange2")
public void sendOrange2(){
String msg = "send routing key = orange.2 msg " + new Date().toString();
rabbitTemplate.convertAndSend(TopicExchangeConfig.TOPIC_EXCHANGE, "orange.2", msg);
}
/**
* 发送路由key为"1.black.1"的消息到Topic交换机，交换机会将该消息投递给队列"queue_black"
*/
@GetMapping("topic/black1")
public void sendBlack1(){
String msg = "send routing key = 1.black msg " + new Date().toString();
rabbitTemplate.convertAndSend(TopicExchangeConfig.TOPIC_EXCHANGE, "1.black.1", msg);
}
/**
* 发送路由key为"2.black.2"的消息到Topic交换机，交换机会将该消息投递给队列"queue_black"
*/
@GetMapping("topic/black2")
public void sendBlack2(){
String msg = "send routing key = 2.black msg " + new Date().toString();
rabbitTemplate.convertAndSend(TopicExchangeConfig.TOPIC_EXCHANGE, "2.black.2", msg);
}
}
3.查看队列
服务启动后，将RabbitMQ上将会看到这两个队列
队列.png
topic交换机.png
三、消费者
1.RabbitMqConfig.java
/**
* 定义队列
*/
@Configuration
public class RabbitMqConfig {
public static final String QUEUE_ORANGE = "queue_orange";
public static final String QUEUE_BLACK = "queue_black";
}
2.RabbitMqReceiver.java
@Component
public class RabbitMqReceiver {
private final static Logger logger = LoggerFactory.getLogger(RabbitMqReceiver.class);
/**
* 监听队列queue_orange的消息
*/
@RabbitListener(queues = RabbitMqConfig.QUEUE_ORANGE)
public void receiverQueueOrange(String msg, Channel channel, Message message) throws IOException {
logger.info("receiverQueueOrange 接收到消息为："+msg);
}
/**
* 监听队列queue_black的消息
*/
@RabbitListener(queues = RabbitMqConfig.QUEUE_BLACK)
public void receiverQueueBlack(String msg, Channel channel, Message message) throws IOException {
logger.info("receiverQueueBlack 接收到消息为："+msg);
}
}
