网站建设完成外网无法访问wordpress 发布模块

网站建设完成外网无法访问,wordpress 发布模块,建网站备案需要的材料,广告代理商是什么意思要实现生产者确认机制失败后自动重试重新投递#xff0c;核心思路是#xff1a;将发送失败的消息暂存→按策略重试→跟踪重试状态→失败兜底。以下是具体实现思路和关键步骤#xff0c;结合代码示例说明。一、核心思路框架当生产者通过 ConfirmCallback 收到 ackfalse#…要实现生产者确认机制失败后自动重试重新投递核心思路是将发送失败的消息暂存→按策略重试→跟踪重试状态→失败兜底。以下是具体实现思路和关键步骤结合代码示例说明。一、核心思路框架当生产者通过 ConfirmCallback 收到 ackfalseBroker 未确认接收或超时未收到确认时说明消息发送失败。此时需将消息暂存到可靠存储避免内存丢失并按重试策略次数、间隔重新投递直至成功或超过阈值后转入死信队列。二、关键实现步骤1. 定义“失败消息”存储结构需持久化存储失败消息的核心信息确保重启后不丢失。推荐用 数据库MySQL/PostgreSQL或 Redis缓存持久化字段包括字段名 说明 示例值msg_id 消息唯一IDCorrelationData的ID MSG-1690000000000-0.123456exchange 目标交换机名称 order.exchangerouting_key 目标路由键 order.routingKeymessage_body 消息体JSON序列化 {id:1001,amount:99.9}retry_count 当前重试次数 0初始值max_retry 最大重试次数如3次 3next_retry_time 下次重试时间时间戳 1690000060000当前时间10秒status 状态待重试/重试中/失败 PENDING2. 实现“失败消息”存储层以 MySQL 为例创建表存储失败消息CREATE TABLE mq_failed_message (id BIGINT PRIMARY KEY AUTO_INCREMENT,msg_id VARCHAR(64) NOT NULL UNIQUE COMMENT 消息唯一ID,exchange VARCHAR(128) NOT NULL COMMENT 交换机,routing_key VARCHAR(128) NOT NULL COMMENT 路由键,message_body TEXT NOT NULL COMMENT 消息体(JSON),retry_count INT DEFAULT 0 COMMENT 当前重试次数,max_retry INT DEFAULT 3 COMMENT 最大重试次数,next_retry_time BIGINT NOT NULL COMMENT 下次重试时间戳(ms),status VARCHAR(20) DEFAULT PENDING COMMENT 状态:PENDING/RETRYING/FAILED,created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,INDEX idx_next_retry_time (next_retry_time) COMMENT 按下次重试时间索引);3. 发送消息时关联“失败存储”生产者发送消息时生成唯一 msg_id如 UUID 或雪花算法并在 ConfirmCallback 中处理失败逻辑Servicepublic class RetryProducer {Autowiredprivate RabbitTemplate rabbitTemplate;Autowiredprivate FailedMessageStorage failedMessageStorage; // 失败消息存储接口public void sendWithRetry(Object message, String exchange, String routingKey) {// 1. 生成唯一消息IDString msgId MSG- System.currentTimeMillis() - UUID.randomUUID().toString().substring(0, 8);CorrelationData correlationData new CorrelationData(msgId);// 2. 发送消息携带CorrelationDatarabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);// 3. 注册确认回调处理成功/失败rabbitTemplate.setConfirmCallback((correlationData1, ack, cause) - {String id correlationData1 ! null ? correlationData1.getId() : null;if (id null) return;if (ack) {// 确认成功删除存储中的失败记录若有failedMessageStorage.deleteById(id);System.out.println(消息确认成功ID: id);} else {// 确认失败存入失败消息表等待重试FailedMessage failedMsg new FailedMessage();failedMsg.setMsgId(id);failedMsg.setExchange(exchange);failedMsg.setRoutingKey(routingKey);failedMsg.setMessageBody(JSON.toJSONString(message)); // 序列化为JSONfailedMsg.setRetryCount(0);failedMsg.setMaxRetry(3);failedMsg.setNextRetryTime(System.currentTimeMillis() 10 * 1000); // 10秒后重试failedMsg.setStatus(PENDING);failedMessageStorage.save(failedMsg);System.err.println(消息确认失败存入重试队列ID: id 原因: cause);}});}}4. 实现重试调度器核心通过 定时任务如 Spring Scheduler或 线程池 定期检查失败消息表对 next_retry_time ≤ 当前时间 的消息执行重试ComponentEnableSchedulingpublic class RetryScheduler {Autowiredprivate FailedMessageStorage failedMessageStorage;Autowiredprivate RabbitTemplate rabbitTemplate;Autowiredprivate DeadLetterQueueHandler deadLetterQueueHandler; // 死信队列处理器// 每5秒扫描一次待重试消息Scheduled(fixedRate 5000)public void retryFailedMessages() {ListFailedMessage pendingMsgs failedMessageStorage.queryPendingMessages(System.currentTimeMillis());for (FailedMessage msg : pendingMsgs) {try {// 1. 标记为“重试中”避免并发重复处理failedMessageStorage.updateStatus(msg.getMsgId(), RETRYING);// 2. 反序列化消息体Object message JSON.parseObject(msg.getMessageBody(), Object.class); // 根据实际类型强转// 3. 重新发送消息携带原msgIdCorrelationData correlationData new CorrelationData(msg.getMsgId());rabbitTemplate.convertAndSend(msg.getExchange(), msg.getRoutingKey(), message, correlationData);// 4. 更新重试次数和下次重试时间指数退避10s→20s→40sint newRetryCount msg.getRetryCount() 1;long nextRetryInterval (long) (10 * Math.pow(2, newRetryCount - 1)) * 1000; // 指数退避long nextRetryTime System.currentTimeMillis() nextRetryInterval;if (newRetryCount msg.getMaxRetry()) {// 超过最大重试次数转入死信队列deadLetterQueueHandler.sendToDlx(msg);failedMessageStorage.updateStatus(msg.getMsgId(), FAILED);System.err.println(消息重试次数耗尽转入死信队列ID: msg.getMsgId());} else {// 更新重试状态次数1下次重试时间failedMessageStorage.updateRetryInfo(msg.getMsgId(), newRetryCount, nextRetryTime, PENDING);System.out.println(消息重试中ID: msg.getMsgId() 第 newRetryCount 次);}} catch (Exception e) {// 重试过程中异常恢复状态为PENDING等待下次扫描failedMessageStorage.updateStatus(msg.getMsgId(), PENDING);System.err.println(重试发送失败ID: msg.getMsgId() 错误: e.getMessage());}}}}5. 死信队列兜底重试失败的最终处理当消息超过最大重试次数仍未成功将其转入死信队列DLX人工介入处理Componentpublic class DeadLetterQueueHandler {Autowiredprivate RabbitTemplate rabbitTemplate;public void sendToDlx(FailedMessage msg) {// 发送到死信交换机需提前配置死信队列和交换机rabbitTemplate.convertAndSend(dlx.exchange,dlx.routingKey,msg.getMessageBody(),message - {MessageProperties props message.getMessageProperties();props.setHeader(failed_reason, max_retry_exceeded);props.setHeader(original_msg_id, msg.getMsgId());return message;});}}三、关键技术点说明1. 重试策略指数退避重试间隔随次数递增如 10s→20s→40s避免集中重试压垮 Broker固定间隔简单场景用固定间隔如每 30 秒重试一次随机间隔避免多个生产者同时重试导致“惊群效应”。2. 幂等性保障重试可能导致消息重复投递消费者需通过 唯一ID去重如 Redis 记录已处理 msg_id有效期 24 小时Componentpublic class IdempotentConsumer {Autowiredprivate RedisTemplateString, Object redisTemplate;RabbitListener(queues order.queue)public void consume(String message, Header(msg_id) String msgId) {String key processed_msg: msgId;if (Boolean.TRUE.equals(redisTemplate.hasKey(key))) {// 已处理直接确认channel.basicAck(deliveryTag, false);return;}// 业务逻辑处理...redisTemplate.opsForValue().set(key, 1, 24, TimeUnit.HOURS); // 记录已处理channel.basicAck(deliveryTag, false);}}3. 存储选型对比存储方式 优点 缺点 适用场景MySQL 持久化可靠支持复杂查询 性能较低需维护数据库连接 生产环境消息重要性高Redis 高性能支持过期时间 数据易失需开启RDB/AOF持久化 中小规模、对性能要求高的场景内存队列 速度快 重启丢失消息 测试环境或临时重试四、总结生产者确认失败后的重试重新投递本质是“存储-调度-重试-兜底”的闭环存储用数据库/Redis 持久化失败消息调度定时任务扫描待重试消息重试按指数退避策略重新发送更新重试状态