【RabbitMQ】rabbitmq广播模式的使用

媒介：
        项目需要同步另一个系统的数据，对方系统接纳MQ的发布/订阅模式方便我们同步数据，即当对方系统中的某条数据修改后，会向绑定他们交换机的每一个队列发布消息。消费者（即我们）监听到消息变动，举行信息消费同步至我们库中。
我们需要做的就是：
        1、创建一个新队列绑定到对方系统的交换机
        2、将监听到的消息举行合明白析，取出消息中的请求头：
              请求头信息为："R"  ，则代表该生为入学操作；
              请求头信息为："X"  ，则代表该生为休学操作；
              请求头信息为："T"  ，则代表该生为退学操作；
        3、接下来根据获取到的请求头内容，来对对方系统传来的数据举行对应操作。
上代码，看思路：
    实现1：

1. /**
2. * @Author: 宁兴星
3. * @CreateTime: 2026-01-16  14:05
4. * @Description: TODO
5. */
6. @Configuration
7. public class RabbitMqConfig extends AbstractRabbitMQConfig {
10.     /**
11.      * 创建广播模式交换机（扇形）
12.      */
13.     @Bean
14.     public FanoutExchange fanoutExchange() {
15.         return new FanoutExchange(EventConstant.STUDENT_EXCHANGE, true, false);
16.     }
18.     /**
19.      * 创建被监听的队列
20.      */
21.     @Bean
22.     public Queue dealerInfoQueue() {
23.         return new Queue(EventConstant.STUDENT_QUEUE, true, false, false);
24.     }
26.     /**
27.      * 将队列绑定到扇形交换机上，实现广播模式消息接收
28.      *
29.      * @param dealerInfoQueue
30.      * @param fanoutExchange
31.      * @return
32.      */
33.     @Bean
34.     public Binding binding(Queue dealerInfoQueue, FanoutExchange fanoutExchange) {
35.         return BindingBuilder.bind(dealerInfoQueue).to(fanoutExchange);
36.     }
38.     /**
39.      * 配置消息监听容器工厂
40.      *
41.      * @param connectionFactory
42.      * @return
43.      */
44.     @Bean
45.     public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
46.         SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
47.         factory.setConnectionFactory(connectionFactory);
48.         factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
49.         factory.setPrefetchCount(10);
50.         return factory;
51.     }
53. }

复制代码

实现2：
       

1.     /**
2.      * MQ监听学生数据变更
3.      *
4.      * @param message 消息体
5.      * @param deliveryTag 消息标识
6.      * @param channel 通道
7.      * @throws IOException IO异常
8.      */
9.     @RabbitListener(queues = EventConstant.STUDENT_QUEUE)
10.     @Operation(summary = "MQ监听学生数据变更", description = "MQ监听学生数据变更")
11.     public void handleMessage(Message message,
12.                             @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
13.                             Channel channel) throws IOException {
14.         try {
15.             log.info("同步学生数据，接收到MQ消息: {}", message);
16.             
17.             // 解析学生信息
18.             StudentInfo studentInfo = parseStudentInfo(message);
19.             log.info("解析后的学生数据: {}", studentInfo);
20.             
21.             // 获取操作类型并处理
22.             String action = getActionFromHeaders(message);
23.             processStudentAction(action, studentInfo);
24.             
25.             // 确认消息处理完成
26.             channel.basicAck(deliveryTag, false);
27.             
28.         } catch (Exception e) {
29.             log.error("处理学生数据消息异常: ", e);
30.             // 消息处理失败,重新入队
31.             channel.basicNack(deliveryTag, false, true);
32.         }
33.     }
35.     /**
36.      * 解析消息中的学生信息
37.      */
38.     private StudentInfo parseStudentInfo(Message message) throws IOException {
39.         ObjectMapper objectMapper = new ObjectMapper();
40.         return objectMapper.readValue(message.getBody(), StudentInfo.class);
41.     }
43.     /**
44.      * 从消息头获取action
45.      */
46.     private String getActionFromHeaders(Message message) {
47.         Map<String, Object> headers = message.getMessageProperties().getHeaders();
48.         String action = headers.get("action").toString();
49.         log.info("操作类型: {}", action);
50.         return action;
51.     }

复制代码

实现3：

1.      /**
2.      * 根据不同action处理学生数据
3.      */
4.     private void processStudentAction(String action, StudentInfo studentInfo) {
5.         if (action == null) {
6.             return;
7.         }
8.         
9.         switch (action) {
10.             case EventConstant.LIGHT_UP:
11.                 // 编写对应录取方法，此处省略具体信息
12.                 handleLightUp(studentInfo);
13.                 break;
14.             case EventConstant.OFFLINE:
15.                 // 编写对应休学方法，此处省略具体信息
16.                 handleOffline(studentInfo);
17.                 break;
18.             case EventConstant.DELETE:
19.                 // 编写对应退学方法，此处省略具体信息
20.                 handleDelete(studentInfo);
21.                 break;
22.             default:
23.                 log.warn("未知的操作类型: {}", action);
24.         }
25.     }

复制代码

竣事啦，如有错误，敬请雅正！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。