SpringBoot实现发布/订阅广播消息的示例代码

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    listener:
      simple:
        # 手动ACK 保证广播消息不丢
        acknowledge-mode: manual
        # 限制预取数，防止单节点消息堆积
        prefetch: 5
        # 开启消费重试
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class FanoutBroadcastConfig {
    // 广播交换机名称
    public static final String FANOUT_EXCHANGE = "system.fanout.broadcast.exchange";
    // 三个独立消费者队列
    public static final String QUEUE_CACHE_REFRESH = "queue.cache.refresh";
    public static final String QUEUE_NOTICE = "queue.system.notice";
    public static final String QUEUE_LOG = "queue.log.collect";
    // 声明 Fanout 广播交换机：持久化、不自动删除
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUT_EXCHANGE, true, false);
    }
    // 队列1：缓存刷新队列
    @Bean
    public Queue cacheRefreshQueue() {
        return new Queue(QUEUE_CACHE_REFRESH, true);
    }
    // 队列2：系统通知队列
    @Bean
    public Queue noticeQueue() {
        return new Queue(QUEUE_NOTICE, true);
    }
    // 队列3：日志采集队列
    @Bean
    public Queue logQueue() {
        return new Queue(QUEUE_LOG, true);
    }
    // 全部绑定到广播交换机
    @Bean
    public Binding bindingCacheRefresh(Queue cacheRefreshQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(cacheRefreshQueue).to(fanoutExchange);
    }
    @Bean
    public Binding bindingNotice(Queue noticeQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(noticeQueue).to(fanoutExchange);
    }
    @Bean
    public Binding bindingLog(Queue logQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(logQueue).to(fanoutExchange);
    }
}

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class BroadcastProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @GetMapping("/send/broadcast")
    public String sendBroadcastMsg(@RequestParam String content) {
        // Fanout广播：路由键传空字符串
        rabbitTemplate.convertAndSend(
                FanoutBroadcastConfig.FANOUT_EXCHANGE,
                "",
                content
        );
        return "✅ 广播消息发送成功：" + content;
    }
}

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
public class CacheRefreshConsumer {
    @RabbitListener(queues = FanoutBroadcastConfig.QUEUE_CACHE_REFRESH)
    public void consume(String msg, Message message, Channel channel) throws IOException {
        try {
            System.out.println("【缓存服务】接收广播消息：" + msg);
            // 执行缓存刷新业务逻辑
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            // 消费失败，重回队列重试
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

@Component
public class SystemNoticeConsumer {
    @RabbitListener(queues = FanoutBroadcastConfig.QUEUE_NOTICE)
    public void consume(String msg, Message message, Channel channel) throws IOException {
        try {
            System.out.println("【通知服务】接收广播消息：" + msg);
            // 执行消息推送业务
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

@Component
public class LogCollectConsumer {
    @RabbitListener(queues = FanoutBroadcastConfig.QUEUE_LOG)
    public void consume(String msg, Message message, Channel channel) throws IOException {
        try {
            System.out.println("【日志服务】接收广播消息：" + msg);
            // 执行日志采集业务
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

http://localhost:8080/send/broadcast?content=全局缓存刷新通知

【缓存服务】接收广播消息：全局缓存刷新通知
【通知服务】接收广播消息：全局缓存刷新通知
【日志服务】接收广播消息：全局缓存刷新通知