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); } }