生产者
配置生产者
/**
* @author BNTang
*/
@Configuration
public class FanoutConfig {
/**
* 声明交换机
*
* @return 交换机
*/
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("logs");
}
/**
* 声明队列
*
* @return 队列
*/
@Bean
public Queue fanoutQueue1() {
return new Queue("fanout_queue1");
}
/**
* 声明队列
*
* @return 队列
*/
@Bean
public Queue fanoutQueue2() {
return new Queue("fanout_queue2");
}
/**
* 把fanout_queue1队列绑定到交换机
*
* @return 交换机
*/
@Bean
public Binding binding1() {
return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
}
/**
* 把fanout_queue2队列绑定到交换机
*
* @return 交换机
*/
@Bean
public Binding binding2() {
return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
}
}
发送消息
@Test
public void testFanout() {
this.rabbitTemplate.convertAndSend("logs", "", "这是日志广播");
System.out.println("消息发送成功");
}
消费者
消费消息
/**
* @author BNTang
*/
@Component
public class FanoutCustomer {
@RabbitListener(bindings = @QueueBinding(
value = @Queue("fanout_queue1"),
exchange = @Exchange(name = "logs", type = ExchangeTypes.FANOUT)
))
public void receive1(String message) {
System.out.println("消费者【1】接收到消息:" + message);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue("fanout_queue2"),
exchange = @Exchange(name = "logs", type = ExchangeTypes.FANOUT)
))
public void receive2(String message) {
System.out.println("消费者【2】接收到消息:" + message);
}
}
以上注解中的属性解释如下:
- value:绑定队列。
- exchange:绑定交换机。
- type:声明交换机类型。
测试方式同之前章节中的一样。