生产者
配置生产者
/**
* @author BNTang
*/
public class RoutingTopicConfig {
/**
* 声明交换机
*
* @return 交换机
*/
public TopicExchange topicExchange() {
return new TopicExchange("topics");
}
/**
* 声明队列1 绑定info和warm
*
* @return 队列1
*/
public Queue topicQueue1() {
return new Queue("topicQueue1");
}
/**
* 声明队列2
*
* @return 队列2
*/
public Queue topicQueue2() {
return new Queue("topicQueue2");
}
/**
* 把队列1 绑定到交换机里面指定user.*的路由key
*
* @return 绑定之后的一个关系
*/
public Binding binding1() {
return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("user.*");
}
/**
* 把队列2 绑定到交换机里面指定user.#的路由key
*
* @return 绑定之后的一个关系
*/
public Binding binding2() {
return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("user.#");
}
}
发送消息
public void testTopic() {
this.rabbitTemplate.convertAndSend("topics", "user.save", "user.save 的消息");
this.rabbitTemplate.convertAndSend("topics", "user.save.findAll", "user.save.findAll 的消息");
this.rabbitTemplate.convertAndSend("topics", "user", "user 的消息");
}
消费者
消费消息
/**
* @author BNTang
*/
public class RoutingTopicConsumer {
(bindings = {
(
value = ("topicQueue1"),
key = {"user.*"},
exchange = (name = "topics", type = ExchangeTypes.TOPIC)
)
})
public void receive1(String message) {
System.out.println("消费者【1】接收到【user.*】消息:" + message);
}
(bindings = {
(
value = ("topicQueue2"),
key = {"user.#"},
exchange = (name = "topics", type = ExchangeTypes.TOPIC)
)
})
public void receive2(String message) {
System.out.println("消费者【2】接收到【user.#】消息:" + message);
}
}
测试方式同之前章节中的一样。