核心内容摘要
《那一刻，全车都跪了：怀孕8个月，她在公交上甩出了顶级撕裂Bass》
基于 RabbitMQ 构建异步化淘客订单处理流水线解耦、削峰与失败重试大家好我是 微赚淘客系统
0 的研发者省赚客在微赚淘客系统
0中用户通过推广链接下单后平台需完成一系列操作验证订单有效性、计算佣金、更新用户收益、发送通知等。
这些操作若同步执行不仅响应慢还容易因第三方接口抖动导致主流程失败。
为此我们基于 RabbitMQ 构建了一套高可用、可扩展的异步订单处理流水线实现服务解耦、流量削峰与自动失败重试。
消息模型设计我们将淘客订单事件抽象为TaobaoOrderEvent通过交换机路由至不同队列按业务阶段分阶段消费。
packagejuwatech.cn.order.event;importjava.io.Serializable;importjava.math.BigDecimal;publicclassTaobaoOrderEventimplementsSerializable{privateStringorderId;// 淘宝订单号privateStringuserId;// 推广用户IDprivateBigDecimalcommission;// 佣金金额元privateStringstatus;// 订单状态VALID / INVALIDprivatelongtimestamp;// getters setters}RabbitMQ 拓扑结构如下Exchange:taoke.order.exchangetopic 类型Queue:taoke.order.process.queueRouting Key:order.process
生产者订单事件发布当接收到淘宝联盟回调时校验签名后立即发布事件不阻塞 HTTP 响应。
packagejuwatech.cn.order.publisher;importjuwatech.cn.order.event.TaobaoOrderEvent;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.stereotype.Component;ComponentpublicclassOrderEventPublisher{privatefinalRabbitTemplaterabbitTemplate;publicOrderEventPublisher(RabbitTemplaterabbitTemplate){this.rabbitTemplaterabbitTemplate;}publicvoidpublish(TaobaoOrderEventevent){rabbitTemplate.convertAndSend(taoke.order.exchange,order.process,event);}}
消费者异步处理流水线消费者监听队列依次执行佣金计算、账户入账、消息通知等逻辑。
packagejuwatech.cn.order.consumer;importjuwatech.cn.order.event.TaobaoOrderEvent;importjuwatech.cn.commission.service.CommissionService;importjuwatech.cn.notify.service.NotifyService;importcom.rabbitmq.client.Channel;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.amqp.support.AmqpHeaders;importorg.springframework.messaging.handler.annotation.Header;importorg.springframework.stereotype.Component;importjava.io.IOException;ComponentpublicclassOrderProcessConsumer{privatefinalCommissionServicecommissionService;privatefinalNotifyServicenotifyService;publicOrderProcessConsumer(CommissionServicecommissionService,NotifyServicenotifyService){this.commissionServicecommissionService;this.notifyServicenotifyService;}RabbitListener(queuestaoke.order.process.queue)publicvoidhandle(TaobaoOrderEventevent,Channelchannel,Header(AmqpHeaders.DELIVERY_TAG)longdeliveryTag)throwsIOException{try{if(VALID.equals(event.getStatus())){commissionService.calculateAndCredit(event.getUserId(),event.getCommission());notifyService.sendEarningsNotice(event.getUserId(),event.getCommission());}channel.basicAck(deliveryTag,false);}catch(Exceptione){// 失败则拒绝并重新入队最多重试3次try{channel.basicNack(deliveryTag,false,true);}catch(IOExceptionex){ex.printStackTrace();}}}}
失败重试与死信队列为避免无限重试导致资源浪费我们配置了重试次数上限并将最终失败消息转入死信队列DLQ供人工处理。
队列声明与 DLQ 绑定packagejuwatech.cn.order.config;importorg.springframework.amqp.core.*;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;ConfigurationpublicclassRabbitMQConfig{publicstaticfinalStringPROCESS_QUEUEtaoke.order.process.queue;publicstaticfinalStringDLQ_QUEUEtaoke.order.dlq;publicstaticfinalStringEXCHANGEtaoke.order.exchange;BeanpublicDirectExchangeorderExchange(){returnnewDirectExchange(EXCHANGE);}BeanpublicQueueprocessQueue(){returnQueueBuilder.durable(PROCESS_QUEUE).withArgument(x-dead-letter-exchange,).withArgument(x-dead-letter-routing-key,DLQ_QUEUE).withArgument(x-message-ttl,
// 消息TTL 10秒.withArgument(x-max-length,
.build();}BeanpublicQueuedlqQueue(){returnQueueBuilder.durable(DLQ_QUEUE).build();}BeanpublicBindingbinding(){returnBindingBuilder.bind(processQueue()).to(orderExchange()).with(order.process);}}
死信消费者告警与人工干预packagejuwatech.cn.order.dlq;importjuwatech.cn.order.event.TaobaoOrderEvent;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;ComponentpublicclassDeadLetterConsumer{privatestaticfinalLoggerlogLoggerFactory.getLogger(DeadLetterConsumer.class);RabbitListener(queuestaoke.order.dlq)publicvoidhandleDeadMessage(TaobaoOrderEventevent){log.error(订单处理最终失败进入死信队列: orderId{}, userId{},event.getOrderId(),event.getUserId());// 触发企业微信/邮件告警AlertService.sendAlert(淘客订单处理失败,event.toString());}}
削峰与批量处理优化在大促期间订单回调量激增。
我们通过以下方式应对增加消费者实例水平扩展OrderProcessConsumer批量确认Batch Ack提升吞吐本地缓存预检避免重复处理同一订单。
// 在 CommissionService 中加入缓存防重privatefinalCacheString,BooleanprocessedCacheCaffeine.newBuilder().expireAfterWrite(24,TimeUnit.HOURS).maximumSize(1_000_
.build();publicvoidcalculateAndCredit(StringuserId,BigDecimalamount){Stringkeycomm_userId_amount;if(processedCache.getIfPresent(key)!null){return;// 已处理}// 执行入账逻辑accountRepo.credit(userId,amount);processedCache.put(key,true);}通过该异步流水线系统成功将订单处理 P99 延迟从
8s 降至 200ms 以内峰值 QPS 支撑能力提升 5 倍且保障了数据最终一致性。
本文著作权归 微赚淘客系统