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_QUEUElucky.danmaku.queue;publicstaticfinalStringDANMAKU_EXCHANGElucky.danmaku.exchange;publicstaticfinalStringDANMAKU_ROUTING_KEYlucky.danmaku.routing;// 抽奖结果三件套publicstaticfinalStringLOTTERY_RESULT_QUEUElucky.lottery.result.queue;publicstaticfinalStringLOTTERY_RESULT_EXCHANGElucky.lottery.result.exchange;publicstaticfinalStringLOTTERY_RESULT_ROUTING_KEYlucky.lottery.result.routing;// 在线人数三件套publicstaticfinalStringONLINE_COUNT_QUEUElucky.online.count.queue;publicstaticfinalStringONLINE_COUNT_EXCHANGElucky.online.count.exchange;publicstaticfinalStringONLINE_COUNT_ROUTING_KEYlucky.online.count.routing;// 死信三件套publicstaticfinalStringDEAD_LETTER_QUEUElucky.dead.letter.queue;publicstaticfinalStringDEAD_LETTER_EXCHANGElucky.dead.letter.exchange;publicstaticfinalStringDEAD_LETTER_ROUTING_KEYlucky.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(){MapString,ObjectargsnewHashMap();// ① 绑定死信交换机这个队列里的消息变成死信时转发到死信交换机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){RabbitTemplatetemplatenewRabbitTemplate(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);// ④ 必须设 trueReturn 回调才会生效returntemplate;}生产者怎么发消息的Slf4jComponentRequiredArgsConstructorpublicclassMessageProducer{privatefinalRabbitTemplaterabbitTemplate;// 注入上面配置的 RabbitTemplatepublicvoidsendDanmaku(MapString,ObjectdanmakuData){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的嵌入OverrideRateLimit(prefixdanmaku_send,maxCount1,windowSeconds1,typeRateLimit.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);// 拒绝记录可以异步DanmakurejectednewDanmaku();rejected.setStatus(2);// 已拒绝returnrejected;}// 异步分界线重活丢给 MQ // 4. 构建消息发送到 RabbitMQ内存操作几毫秒MapString,ObjectdanmakuMessagebuildDanmakuMessage(dto);messageProducer.sendDanmaku(danmakuMessage);// 5. 直接返回成功不等推送、不等写库DanmakuresultnewDanmaku();result.setActivityId(dto.getActivityId());result.setParticipantId(dto.getParticipantId());result.setContent(dto.getContent());result.setStatus(1);// 自动通过returnresult;}比较轻量的业务逻辑就在同步完成涉及到内存的这些比较重的就在异步完成消费者怎么处理消息的包含手动ACK、场次定向推送、批量入库、有限重试、死信兜底RabbitListener(queuesRabbitConfig.DANMAKU_QUEUE)// ① 监听弹幕队列publicvoidhandleDanmaku(MapString,ObjectdanmakuData,Channelchannel,// ② Channel用于手动 ACKHeader(AmqpHeaders.DELIVERY_TAG)longdeliveryTag)throwsIOException{StringretryKeyMQ_RETRY_KEY_PREFIXdanmaku:deliveryTag;// ③ 重试计数的 Redis keytry{// 1. 解析 activityId用于场次定向推送LongactivityIdnull;ObjectactivityIdObjdanmakuData.get(FIELD_ACTIVITY_ID);if(activityIdObj!null){activityIdLong.parseLong(activityIdObj.toString());}// 2. 场次定向批量推送核心优化只推当前活动场次不全局广播if(activityId!null){webSocketHandler.batchBroadcastDanmaku(activityId,danmaku,danmakuData);}else{webSocketHandler.broadcast(danmaku,danmakuData);}// 3. 异步批量入库不阻塞推送ObjectstatusObjdanmakuData.get(FIELD_STATUS);intstatusstatusObj!null?Integer.parseInt(statusObj.toString()):1;DanmakudanmakunewDanmaku();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的弹幕批量入库Slf4jComponentRequiredArgsConstructorpublicclassDanmakuBatchPersister{privatefinalDanmakuMapperdanmakuMapper;// ① 线程安全的缓冲队列无界链表队列privatefinalConcurrentLinkedQueueDanmakubuffernewConcurrentLinkedQueue();privatestaticfinalintBATCH_SIZE100;// ② 批量阈值100条// ③ 提交一条弹幕到缓冲publicvoidsubmit(Danmakudanmaku){if(danmaku.getCreatedAt()null){danmaku.setCreatedAt(LocalDateTime.now());}buffer.offer(danmaku);// 加入队列尾部// ④ 达到100条立即刷新数量触发if(buffer.size()BATCH_SIZE){flush();}}// ⑤ 定时刷新每100ms检查一次有数据就刷时间触发Scheduled(fixedRate100)publicvoidscheduledFlush(){if(!buffer.isEmpty()){flush();}}// ⑥ 执行批量入库privatevoidflush(){ListDanmakubatchnewArrayList(BATCH_SIZE);Danmakuitem;// 从缓冲队列取出最多100条while((itembuffer.poll())!nullbatch.size()BATCH_SIZE){batch.add(item);}if(batch.isEmpty())return;try{// 逐条插入MyBatis-Plus 没有真正的批量insert这里是循环insertfor(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就检查有数据就flushPreDestroy是在停机之前把缓冲东西写到内存里面到目前为止的全链路═══════════════════════════════════════════════════════════════ 【同步主链路】用户发弹幕请求 → 立即返回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 → current1 ≤ maxCount1 → 放行 ⑤ 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.exchangeroutingKeylucky.danmaku.routing → Direct 交换机精确匹配 → 路由到 lucky.danmaku.queue → Publisher Confirm 回调acktrue消息到达交换机 ⑦ 返回 Result.ok(danmaku) → 前端收到发送成功 ⚠️ 此时弹幕还没上大屏、还没写库但用户已经看到发送成功了 ═══════════════════════════════════════════════════════════════ 【异步消费链路】MQ 消费者取消息 → 推送 入库消费者按能力匀速处理 ═══════════════════════════════════════════════════════════════ ⑧ MessageConsumer.handleDanmaku() 被 RabbitListener 触发4个并发线程之一 → 从 lucky.danmaku.queue 取出消息prefetch10一次最多拉10条 → deliveryTag某编号 → retryKey lucky:mq:retry:danmaku: deliveryTag ⑨ 解析 activityId1 ⑩ webSocketHandler.batchBroadcastDanmaku(1, danmaku, message) → 从 activitySessionMap 拿到 activityId1 的所有在线 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 → retryCount1 → 1 ≤ 3 → channel.basicNack(deliveryTag, false, true) → requeuetrue → 消息重新放回队列头部等待下一次消费 ⑭ 第2次、第3次消费还是失败 → 同样 requeue ⑮ 第4次消费失败 → retryCount4 3 → channel.basicNack(deliveryTag, false, false) → requeuefalse → 消息变成死信 → 转发到 lucky.dead.letter.exchange → 路由到 lucky.dead.letter.queue 存起来 → 运维人员在 RabbitMQ 管理后台看到死信排查原因回答一下我的疑问这么多交换机我一跳消息到MQ里面后怎么知道去哪一个交换机生产者发消息的时候指定了交换机