application.yml这里面的代码

rabbitmq:host:${RABBITMQ_HOST:localhost}#RabbitMQ地址（Docker里是 lucky-rabbitmq） port:${RABBITMQ_PORT:5672}# 端口：5672是AMQP协议端口，15672是管理后台 username:${RABBITMQ_USERNAME:guest}password:${RABBITMQ_PASSWORD:guest}virtual-host:/# 虚拟主机（类似数据库的 schema，隔离不同业务） #=====生产者可靠性配置=====publisher-confirm-type:correlated # ① 开启发布确认：消息到达交换机后回调ACK/NACKpublisher-returns:true# ② 开启发布返回：消息路由不到队列时退回给生产者 #=====消费者配置=====listener:simple:acknowledge-mode:manual # ③ 手动ACK：消费者处理完显式确认，MQ才删消息 prefetch:10# ④ 预取数：一次最多从队列拉10条消息，防止消费者被压垮 concurrency:4# ⑤ 消费者并发数：4个线程同时消费，提高吞吐

RabbitConfig中定义交换机、队列、死信的名字

// 弹幕三件套publicstaticfinalStringDANMAKU_QUEUE="lucky.danmaku.queue";publicstaticfinalStringDANMAKU_EXCHANGE="lucky.danmaku.exchange";publicstaticfinalStringDANMAKU_ROUTING_KEY="lucky.danmaku.routing";// 抽奖结果三件套publicstaticfinalStringLOTTERY_RESULT_QUEUE="lucky.lottery.result.queue";publicstaticfinalStringLOTTERY_RESULT_EXCHANGE="lucky.lottery.result.exchange";publicstaticfinalStringLOTTERY_RESULT_ROUTING_KEY="lucky.lottery.result.routing";// 在线人数三件套publicstaticfinalStringONLINE_COUNT_QUEUE="lucky.online.count.queue";publicstaticfinalStringONLINE_COUNT_EXCHANGE="lucky.online.count.exchange";publicstaticfinalStringONLINE_COUNT_ROUTING_KEY="lucky.online.count.routing";// 死信三件套publicstaticfinalStringDEAD_LETTER_QUEUE="lucky.dead.letter.queue";publicstaticfinalStringDEAD_LETTER_EXCHANGE="lucky.dead.letter.exchange";publicstaticfinalStringDEAD_LETTER_ROUTING_KEY="lucky.dead.letter.routing";

• 每一个业务都有一个独立的交换机
• 分开交换机更清晰，流量完全隔离。弹幕量最大，抽奖结果量小但重要，分开后即使弹幕队列积压，也不影响抽奖结果的消费

RabbitConfig死信交换机和队列和路由键的绑定

@BeanpublicQueuedeadLetterQueue(){returnQueueBuilder.durable(DEAD_LETTER_QUEUE).build();// durable=持久化}@BeanpublicDirectExchangedeadLetterExchange(){returnnewDirectExchange(DEAD_LETTER_EXCHANGE);// Direct 类型交换机}@BeanpublicBindingdeadLetterBinding(QueuedeadLetterQueue,DirectExchangedeadLetterExchange){returnBindingBuilder.bind(deadLetterQueue)// 把死信队列.to(deadLetterExchange)// 绑定到死信交换机.with(DEAD_LETTER_ROUTING_KEY);// 用这个 routingKey}

业务队列的定义和死信绑定

@BeanpublicQueuedanmakuQueue(){Map<String,Object>args=newHashMap<>();// ① 绑定死信交换机：这个队列里的消息变成死信时，转发到死信交换机args.put("x-dead-letter-exchange",DEAD_LETTER_EXCHANGE);// ② 死信的 routingKey：转发到死信交换机时用这个 keyargs.put("x-dead-letter-routing-key",DEAD_LETTER_ROUTING_KEY);// ③ 消息 TTL：队列里的消息 30 秒没被消费就过期，变成死信args.put("x-message-ttl",30000);returnQueueBuilder.durable(DANMAKU_QUEUE).withArguments(args).build();}@BeanpublicDirectExchangedanmakuExchange(){returnnewDirectExchange(DANMAKU_EXCHANGE);}@BeanpublicBindingdanmakuBinding(QueuedanmakuQueue,DirectExchangedanmakuExchange){returnBindingBuilder.bind(danmakuQueue).to(danmakuExchange).with(DANMAKU_ROUTING_KEY);}

RabbitTemplate配置回调

@BeanpublicRabbitTemplaterabbitTemplate(ConnectionFactoryconnectionFactory){RabbitTemplatetemplate=newRabbitTemplate(connectionFactory);template.setMessageConverter(jsonMessageConverter());// ① JSON 序列化// ② Publisher Confirm 回调：消息到达交换机后触发template.setConfirmCallback((correlationData,ack,cause)->{if(!ack){log.error("消息发送失败: correlationData={}, cause={}",correlationData,cause);}else{log.debug("消息发送成功: correlationData={}",correlationData);}});// ③ Publisher Return 回调：消息路由不到队列时触发template.setReturnsCallback(returned->{log.error("消息退回: exchange={}, routingKey={}, replyText={}, message={}",returned.getExchange(),returned.getRoutingKey(),returned.getReplyText(),returned.getMessage());});template.setMandatory(true);// ④ 必须设 true，Return 回调才会生效returntemplate;}

生产者怎么发消息的

@Slf4j@Component@RequiredArgsConstructorpublicclassMessageProducer{privatefinalRabbitTemplaterabbitTemplate;// 注入上面配置的 RabbitTemplatepublicvoidsendDanmaku(Map<String,Object>danmakuData){try{rabbitTemplate.convertAndSend(RabbitConfig.DANMAKU_EXCHANGE,// ① 发到哪个交换机RabbitConfig.DANMAKU_ROUTING_KEY,// ② 带什么 routingKeydanmakuData// ③ 消息体（自动 JSON 序列化）);log.debug("弹幕消息已发送到队列: {}",danmakuData.get("content"));}catch(Exceptione){log.error("发送弹幕消息失败: {}",e.getMessage(),e);}}// 抽奖结果、在线人数方法结构完全一样，只是交换机和 routingKey 不同publicvoidsendLotteryResult(Objectwinners){...}publicvoidsendOnlineCount(intcount){...}}

• converAndSend来处理交换机，路由，消息体的东西

业务层中MQ的嵌入

@Override@RateLimit(prefix="danmaku_send",maxCount=1,windowSeconds=1,type=RateLimit.LimitType.CUSTOM,keyExpression="#dto.participantId",message="弹幕发送过于频繁，请稍后再试（每秒最多1条）")publicDanmakusendDanmaku(DanmakuDTOdto){// ===== 同步主链路：必须在返回前完成的轻量操作 =====// 1. 参数校验（内存操作，极快）if(dto.getActivityId()==null)thrownewBusinessException("活动不存在");if(dto.getContent()==null||dto.getContent().trim().isEmpty())thrownewBusinessException("弹幕内容不能为空");// 2. 参与者状态校验（查 Redis，内存操作，极快）if(dto.getParticipantId()!=null){checkParticipantStatus(dto.getParticipantId());// 禁言/移出检查}// 3. 敏感词过滤（内存操作，必须同步——违规内容不能发出去）if(sensitiveWordService.containsSensitiveWord(dto.getContent())){asyncRecordRejectedDanmaku(dto);// 拒绝记录可以异步Danmakurejected=newDanmaku();rejected.setStatus(2);// 已拒绝returnrejected;}// ===== 异步分界线：重活丢给 MQ =====// 4. 构建消息，发送到 RabbitMQ（内存操作，几毫秒）Map<String,Object>danmakuMessage=buildDanmakuMessage(dto);messageProducer.sendDanmaku(danmakuMessage);// 5. 直接返回成功（不等推送、不等写库）Danmakuresult=newDanmaku();result.setActivityId(dto.getActivityId());result.setParticipantId(dto.getParticipantId());result.setContent(dto.getContent());result.setStatus(1);// 自动通过returnresult;}

• 比较轻量的业务逻辑就在同步完成
• 涉及到内存的这些比较重的，就在异步完成

消费者怎么处理消息的

• 包含手动ACK、场次定向推送、批量入库、有限重试、死信兜底

@RabbitListener(queues=RabbitConfig.DANMAKU_QUEUE)// ① 监听弹幕队列publicvoidhandleDanmaku(Map<String,Object>danmakuData,Channelchannel,// ② Channel：用于手动 ACK@Header(AmqpHeaders.DELIVERY_TAG)longdeliveryTag)throwsIOException{StringretryKey=MQ_RETRY_KEY_PREFIX+"danmaku:"+deliveryTag;// ③ 重试计数的 Redis keytry{// 1. 解析 activityId（用于场次定向推送）LongactivityId=null;ObjectactivityIdObj=danmakuData.get(FIELD_ACTIVITY_ID);if(activityIdObj!=null){activityId=Long.parseLong(activityIdObj.toString());}// 2. 场次定向批量推送（核心优化：只推当前活动场次，不全局广播）if(activityId!=null){webSocketHandler.batchBroadcastDanmaku(activityId,"danmaku",danmakuData);}else{webSocketHandler.broadcast("danmaku",danmakuData);}// 3. 异步批量入库（不阻塞推送）ObjectstatusObj=danmakuData.get(FIELD_STATUS);intstatus=statusObj!=null?Integer.parseInt(statusObj.toString()):1;Danmakudanmaku=newDanmaku();danmaku.setActivityId(activityId);// ... 设置其他字段danmakuBatchPersister.submit(danmaku);// ④ 丢进批量缓冲，不直接写库// 4. 手动确认消息（处理成功，告诉 MQ 可以删了）channel.basicAck(deliveryTag,false);redisUtil.delete(retryKey);// ⑤ 成功后清除重试计数log.debug("弹幕消息消费成功: activityId={}, content={}",activityId,danmakuData.get("content"));}catch(Exceptione){// ⑥ 处理失败，走有限重试逻辑handleConsumeFailure(channel,deliveryTag,retryKey,"弹幕",e);}}

• Channel channel是有关手动ACK的
• @Header是消息编号，因为涉及到ACK
• 这个有限重试是写一个表示位在redis里面

DanmakuBatchPersister的弹幕批量入库

@Slf4j@Component@RequiredArgsConstructorpublicclassDanmakuBatchPersister{privatefinalDanmakuMapperdanmakuMapper;// ① 线程安全的缓冲队列（无界链表队列）privatefinalConcurrentLinkedQueue<Danmaku>buffer=newConcurrentLinkedQueue<>();privatestaticfinalintBATCH_SIZE=100;// ② 批量阈值：100条// ③ 提交一条弹幕到缓冲publicvoidsubmit(Danmakudanmaku){if(danmaku.getCreatedAt()==null){danmaku.setCreatedAt(LocalDateTime.now());}buffer.offer(danmaku);// 加入队列尾部// ④ 达到100条立即刷新（数量触发）if(buffer.size()>=BATCH_SIZE){flush();}}// ⑤ 定时刷新：每100ms检查一次，有数据就刷（时间触发）@Scheduled(fixedRate=100)publicvoidscheduledFlush(){if(!buffer.isEmpty()){flush();}}// ⑥ 执行批量入库privatevoidflush(){List<Danmaku>batch=newArrayList<>(BATCH_SIZE);Danmakuitem;// 从缓冲队列取出最多100条while((item=buffer.poll())!=null&&batch.size()<BATCH_SIZE){batch.add(item);}if(batch.isEmpty())return;try{// 逐条插入（MyBatis-Plus 没有真正的批量insert，这里是循环insert）for(Danmakudanmaku:batch){danmakuMapper.insert(danmaku);}log.debug("弹幕批量入库完成: {} 条",batch.size());}catch(Exceptione){log.error("弹幕批量入库失败: {} 条",batch.size(),e);// ⑦ 失败重试：每条最多重试3次，超限丢弃for(Danmakudanmaku:batch){if(danmaku.getRetryCount()==null)danmaku.setRetryCount(0);if(danmaku.getRetryCount()<3){danmaku.setRetryCount(danmaku.getRetryCount()+1);buffer.offer(danmaku);// 重新入队}else{log.error("弹幕入库重试超限，丢弃: content={}",danmaku.getContent());}}}}// ⑧ 应用关闭时刷新剩余缓冲（优雅停机）@PreDestroypublicvoiddestroy(){if(!buffer.isEmpty()){flush();log.info("应用关闭，弹幕批量入库缓冲已刷新");}}}

• 双触发机制
• 缓存到100就flush
• 每100ms就检查，有数据就flush
• @PreDestroy是在停机之前把缓冲东西写到内存里面

到目前为止的全链路

═══════════════════════════════════════════════════════════════ 【同步主链路】用户发弹幕请求 → 立即返回（<10ms） ═══════════════════════════════════════════════════════════════ ① 前端 POST /api/danmaku/send {"activityId":1,"participantId":123,"content":"大家好"} ② JwtAuthenticationFilter → SecurityConfig 授权（/api/danmaku/send 是 permitAll）→ 放行 ③ DanmakuController.sendDanmaku() → DanmakuServiceImpl.sendDanmaku(dto) ④ @RateLimit 切面拦截（第三层）： → Lua 脚本 INCR rate_limit:danmaku_send:123 → current=1 ≤ maxCount=1 → 放行 ⑤ sendDanmaku() 同步逻辑： → 参数校验通过 → checkParticipantStatus(123)：查 Redis，未禁言 → sensitiveWordService.containsSensitiveWord("大家好")：无敏感词 → buildDanmakuMessage(dto)：构建 Map {activityId:1, participantId:123, content:"大家好", status:1, ...} ⑥ messageProducer.sendDanmaku(danmakuMessage)： → rabbitTemplate.convertAndSend("lucky.danmaku.exchange", "lucky.danmaku.routing", message) → Jackson 把 Map 序列化成 JSON → 消息发到 lucky.danmaku.exchange，routingKey=lucky.danmaku.routing → Direct 交换机精确匹配 → 路由到 lucky.danmaku.queue → Publisher Confirm 回调：ack=true（消息到达交换机） ⑦ 返回 Result.ok(danmaku) → 前端收到"发送成功" ⚠️ 此时弹幕还没上大屏、还没写库，但用户已经看到"发送成功"了 ═══════════════════════════════════════════════════════════════ 【异步消费链路】MQ 消费者取消息 → 推送 + 入库（消费者按能力匀速处理） ═══════════════════════════════════════════════════════════════ ⑧ MessageConsumer.handleDanmaku() 被 @RabbitListener 触发（4个并发线程之一）： → 从 lucky.danmaku.queue 取出消息（prefetch=10，一次最多拉10条） → deliveryTag=某编号 → retryKey = "lucky:mq:retry:danmaku:" + deliveryTag ⑨ 解析 activityId=1 ⑩ webSocketHandler.batchBroadcastDanmaku(1, "danmaku", message)： → 从 activitySessionMap 拿到 activityId=1 的所有在线 session → 把消息加入该场次的批量缓冲（batchBuffer） → 50ms定时刷新时，把缓冲里的弹幕合并成一条 JSON，批量推送给该场次所有在线客户端 → 大屏端和移动端收到 WebSocket 消息，弹幕上屏滚动 ✅ ⑪ danmakuBatchPersister.submit(danmaku)： → 弹幕对象加入 ConcurrentLinkedQueue 缓冲 → 缓冲没到100条，不立即写库 → 100ms后 @Scheduled 定时触发 flush() → 取出缓冲里所有弹幕，循环 danmakuMapper.insert() 写入 MySQL ✅ ⑫ channel.basicAck(deliveryTag, false)： → 手动确认，MQ 从 lucky.danmaku.queue 删除这条消息 → redisUtil.delete(retryKey) 清理重试计数 ═══════════════════════════════════════════════════════════════ 【异常分支】如果第⑩步推送失败 ═══════════════════════════════════════════════════════════════ ⑬ catch (Exception e) → handleConsumeFailure()： → Redis INCR retryKey → retryCount=1 → 1 ≤ 3 → channel.basicNack(deliveryTag, false, true) → requeue=true → 消息重新放回队列头部，等待下一次消费 ⑭ 第2次、第3次消费还是失败 → 同样 requeue ⑮ 第4次消费失败： → retryCount=4 > 3 → channel.basicNack(deliveryTag, false, false) → requeue=false → 消息变成死信 → 转发到 lucky.dead.letter.exchange → 路由到 lucky.dead.letter.queue 存起来 → 运维人员在 RabbitMQ 管理后台看到死信，排查原因

回答一下我的疑问

• 这么多交换机，我一跳消息到MQ里面后怎么知道去哪一个交换机？
• 生产者发消息的时候指定了交换机