From cfe9d0e9947f91a537f929eb7b3e9034810b08e1 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Tue, 14 Jul 2026 13:04:28 +0100 Subject: [PATCH] =?UTF-8?q?style(driver):=20=E4=B8=A4=20driver=20=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E6=9C=9F=E6=97=A5=E5=BF=97=E4=B8=8E=E5=BC=82=E5=B8=B8?= =?UTF-8?q?=E6=96=87=E6=A1=88=E4=B8=AD=E6=96=87=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 纯文案改动,无逻辑变化;关键消费路径日志(毒消息 ACK/InProgress 留投/重试耗尽) 统一为中文口径,便于业务侧排查。 Co-Authored-By: Claude Fable 5 --- .../njcn/mq/driver/redis/RedisStreamMqDriver.java | 12 ++++++------ .../com/njcn/mq/driver/rocketmq/RocketMqDriver.java | 10 +++++----- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java index c3da33b..b071b0d 100644 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java @@ -79,7 +79,7 @@ public class RedisStreamMqDriver implements MqDriver { } } catch (Exception e) { if (!running) break; - log.error("[mq-redis] loop error stream={} group={}", stream, group, e); + log.error("[mq-redis] 消费轮询异常 stream={} group={}", stream, group, e); sleep(props.getBlockMs()); } } @@ -92,7 +92,7 @@ public class RedisStreamMqDriver implements MqDriver { String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY)); payload = JSON.parseObject(json, sub.getPayloadType()); } catch (Exception poison) { - log.error("[mq-redis] poison message acked, key={}", fields.get(StreamMessageConstant.F_KEY), poison); + log.error("[mq-redis] 消息反序列化失败，按毒消息 ACK 丢弃 key={}", fields.get(StreamMessageConstant.F_KEY), poison); return true; // 毒消息直接 ACK，不卡队列 } MqMessage msg = MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload); @@ -102,10 +102,10 @@ public class RedisStreamMqDriver implements MqDriver { } catch (MqMessageInProgressException inProgress) { // 去重标记仍 processing（并发处理中/上次崩溃残留）：不 ACK，留 PEL， // 待 processing 标记过期后由 reclaimPending 重新 acquire 并真正重投。 - log.debug("[mq-redis] in-progress, leave PEL key={}", msg.getKey()); + log.debug("[mq-redis] 消息仍在处理中，暂不 ACK、留 PEL 待重投 key={}", msg.getKey()); return false; } catch (Exception e) { - log.warn("[mq-redis] handle failed key={}, leave PEL for retry (max={})", msg.getKey(), maxRetry, e); + log.warn("[mq-redis] 消费失败，留 PEL 待重投 key={} maxRetry={}", msg.getKey(), maxRetry, e); return false; // 留 PEL，由 reclaimPending 重投或落库 } } @@ -142,11 +142,11 @@ public class RedisStreamMqDriver implements MqDriver { try { sub.getRetryExhaustedHandler().accept(buildMessage(sub, fields)); } catch (Exception ex) { - log.error("[mq-redis] retryExhausted handler error key={}", fields.get(StreamMessageConstant.F_KEY), ex); + log.error("[mq-redis] 重试耗尽落库回调执行异常 key={}（不影响主流程）", fields.get(StreamMessageConstant.F_KEY), ex); } } redis.opsForStream().acknowledge(stream, group, rec.getId()); - log.warn("[mq-redis] retry exhausted, acked key={} delivery={}", fields.get(StreamMessageConstant.F_KEY), deliveryCount); + log.warn("[mq-redis] 重试耗尽，已 ACK 终止 key={} delivery={}", fields.get(StreamMessageConstant.F_KEY), deliveryCount); } else { boolean ack = dispatch(sub, fields, maxRetry); if (ack) redis.opsForStream().acknowledge(stream, group, rec.getId()); diff --git a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java index 12dce2e..749b5cc 100644 --- a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java +++ b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java @@ -54,7 +54,7 @@ public class RocketMqDriver implements MqDriver { consumer.start(); consumers.add(consumer); } catch (Exception e) { - throw new IllegalStateException("rocket subscribe failed: " + sub.getTopic(), e); + throw new IllegalStateException("RocketMQ 订阅失败 topic=" + sub.getTopic(), e); } } @@ -67,7 +67,7 @@ public class RocketMqDriver implements MqDriver { } catch (MqMessageInProgressException inProgress) { // 去重"处理中"信号:不 ACK,交 broker 重投;绝不进耗尽落库分支(否则并发下静默丢消息)。 // 语义对齐 RedisStreamMqDriver#dispatch 对该异常的 return false(留 PEL)。 - log.debug("[mq-rocket] in-progress, reconsume later key={}", ext.getKeys()); + log.debug("[mq-rocket] 消息仍在处理中，返回 RECONSUME_LATER 待重投 key={}", ext.getKeys()); return ConsumeConcurrentlyStatus.RECONSUME_LATER; } catch (Exception e) { if (ext.getReconsumeTimes() >= maxRetry) { @@ -75,7 +75,7 @@ public class RocketMqDriver implements MqDriver { //（对齐 redis-stream reclaim 耗尽语义,兑现"换 MQ 业务不动"契约)。 onRetryExhausted(sub, ext, e); } else { - log.warn("[mq-rocket] handle failed msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), e); + log.warn("[mq-rocket] 消费失败，交 broker 重投 msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), e); return ConsumeConcurrentlyStatus.RECONSUME_LATER; } } @@ -85,7 +85,7 @@ public class RocketMqDriver implements MqDriver { /** 重试耗尽落库(best-effort:毒消息时 payload 为 null，至少保留 key/tag),并隔离 handler 自身异常。 */ private void onRetryExhausted(MqSubscription sub, MessageExt ext, Exception cause) { - log.warn("[mq-rocket] retry exhausted, acked msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), cause); + log.warn("[mq-rocket] 重试耗尽，已 ACK 终止 msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), cause); Consumer handler = sub.getRetryExhaustedHandler(); if (handler == null) return; Object payload = null; @@ -97,7 +97,7 @@ public class RocketMqDriver implements MqDriver { try { handler.accept(MqMessage.of(ext.getKeys(), ext.getTags(), payload)); } catch (Exception ex) { - log.error("[mq-rocket] retryExhausted handler error msgId={}（不影响主流程)", ext.getMsgId(), ex); + log.error("[mq-rocket] 重试耗尽落库回调执行异常 msgId={}（不影响主流程）", ext.getMsgId(), ex); } }