rabbitmq实现向各服务广播消息

时间:2023-03-08 19:54:42

广播fanout

主要是将一个消息,分发到绑定了它的队列上,而这些队列如消费者自己去建立和绑定!

对生产者是解耦的

生产者不需要关心消费者有多少,消费者如果需要这种消息,只需要把队列绑定到exchange上即可

流程

  1. 打开rabbitmq的ui
  2. 建立两个队列fanout1,fanout2
  3. 打开exchange里的amqp.fanout类型
  4. 绑定上面的两个队列
  5. 向exchange里发消息
  6. 回到队列页面,这时可以看到每个队列都收到了消息

例子

@Component
public class AmqpConfig {
public static final String LIND_FANOUT_EXCHANGE = "lindFanoutExchange"; /**
* 广播交换机.
*
* @return
*/
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(LIND_FANOUT_EXCHANGE);
}
}

生产者

  /**
* 发布广播消息.
*
* @param message .
*/
public void fanoutPublish(String message) {
try {
rabbitTemplate.convertAndSend(AmqpConfig.LIND_FANOUT_EXCHANGE, null, "广播消息");
} catch (Exception e) {
e.printStackTrace();
}
}

消费者

@Component
public class FanoutSubscriber { @Autowired
AmqpConfig amqpConfig; @Bean
public Queue product1Queue() {
return new Queue("product1.queue");
} @Bean
public Queue product2Queue() {
return new Queue("product2.queue");
} @Bean
public Binding product1QueueBinding() {
return BindingBuilder.bind(product1Queue()).to(amqpConfig.fanoutExchange());
} @Bean
public Binding product2QueueBinding() {
return BindingBuilder.bind(product2Queue()).to(amqpConfig.fanoutExchange());
} @RabbitListener(queues = "product1.queue")
public void product1(String data) {
System.out.println(data);
} @RabbitListener(queues = "product2.queue")
public void product2(String data) {
System.out.println(data);
}
}