[消費(fèi)者接收消息]  
        │  
        ▼  
[解析消息，生成唯一冪等標(biāo)識]  
        │  
        ▼  
[查詢冪等標(biāo)識狀態(tài)]  
        │  
┌───────┴───────┐  
│ 存在且已消費(fèi)  │           [返回成功，丟棄消息]  
└───────┬───────┘  
        │  
┌───────┴───────┐  
│ 存在且消費(fèi)中  │           [延遲消費(fèi)，等待重試]  
└───────┬───────┘  
        │  
┌───────┴───────┐  
│   不存在      │  
└───────┬───────┘  
        │  
[設(shè)置冪等標(biāo)識為“消費(fèi)中”，并設(shè)置過期時(shí)間]  
        │  
        ▼  
[執(zhí)行業(yè)務(wù)邏輯]  
        │  
        ▼  
[業(yè)務(wù)執(zhí)行成功？]  
        │  
┌───────┴───────┐  
│     是        │           [更新標(biāo)識為“已消費(fèi)”]  
│               │           [刪除或保留標(biāo)識]  
└───────┬───────┘  
        │  
┌───────┴───────┐  
│     否        │           [刪除標(biāo)識，允許重試]  
└───────┬───────┘  
        │  
        ▼  
[流程結(jié)束]  

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface NoMQDuplicateConsume {

    /**
     * 設(shè)置防重令牌 Key 前綴
     */
    String keyPrefix() default "";

    /**
     * 通過 SpEL 表達(dá)式生成的唯一 Key
     */
    String key();

    /**
     * 設(shè)置防重令牌 Key 過期時(shí)間，單位秒，默認(rèn) 1 小時(shí)
     */
    long keyTimeout() default 3600L;
}

@RequiredArgsConstructor
public enum IdempotentMQConsumeStatusEnum {

    /**
     * 消費(fèi)中
     */
    CONSUMING("0"),

    /**
     * 已消費(fèi)
     */
    CONSUMED("1");

    @Getter
    private final String code;

    /**
     * 如果消費(fèi)狀態(tài)等于消費(fèi)中，返回失敗
     *
     * @param consumeStatus 消費(fèi)狀態(tài)
     * @return 是否消費(fèi)失敗
     */
    public static boolean isError(String consumeStatus) {
        return Objects.equals(CONSUMING.code, consumeStatus);
    }
}

@Slf4j
@Aspect
@RequiredArgsConstructor
public final class NoMQDuplicateConsumeAspect {

    private final StringRedisTemplate stringRedisTemplate;

    private static final String LUA_SCRIPT = """
            local key = KEYS[1]
            local value = ARGV[1]
            local expire_time_ms = ARGV[2]
            return redis.call('SET', key, value, 'NX', 'GET', 'PX', expire_time_ms)
            """;

    /**
     * 增強(qiáng)方法標(biāo)記 {@link NoMQDuplicateConsume} 注解邏輯
     */
    @Around("@annotation(com.nageoffer.onecoupon.framework.idempotent.NoMQDuplicateConsume)")
    public Object noMQRepeatConsume(ProceedingJoinPoint joinPoint) throws Throwable {
        NoMQDuplicateConsume noMQDuplicateConsume = getNoMQDuplicateConsumeAnnotation(joinPoint);
        String uniqueKey = noMQDuplicateConsume.keyPrefix() + SpELUtil.parseKey(noMQDuplicateConsume.key(), ((MethodSignature) joinPoint.getSignature()).getMethod(), joinPoint.getArgs());

        String absentAndGet = stringRedisTemplate.execute(
                RedisScript.of(LUA_SCRIPT, String.class),
                List.of(uniqueKey),
                IdempotentMQConsumeStatusEnum.CONSUMING.getCode(),
                String.valueOf(TimeUnit.SECONDS.toMillis(noMQDuplicateConsume.keyTimeout()))
        );

        // 如果不為空證明已經(jīng)有
        if (Objects.nonNull(absentAndGet)) {
            boolean errorFlag = IdempotentMQConsumeStatusEnum.isError(absentAndGet);
            log.warn("[{}] MQ repeated consumption, {}.", uniqueKey, errorFlag ? "Wait for the client to delay consumption" : "Status is completed");
            if (errorFlag) {
                throw new ServiceException(String.format("消息消費(fèi)者冪等異常，冪等標(biāo)識：%s", uniqueKey));
            }
            return null;
        }

        Object result;
        try {
            // 執(zhí)行標(biāo)記了消息隊(duì)列防重復(fù)消費(fèi)注解的方法原邏輯
            result = joinPoint.proceed();

            // 設(shè)置防重令牌 Key 過期時(shí)間，單位秒
            stringRedisTemplate.opsForValue().set(uniqueKey, IdempotentMQConsumeStatusEnum.CONSUMED.getCode(), noMQDuplicateConsume.keyTimeout(), TimeUnit.SECONDS);
        } catch (Throwable ex) {
            // 刪除冪等 Key，讓消息隊(duì)列消費(fèi)者重試邏輯進(jìn)行重新消費(fèi)
            stringRedisTemplate.delete(uniqueKey);
            throw ex;
        }
        return result;
    }

    /**
     * @return 返回自定義防重復(fù)消費(fèi)注解
     */
    public static NoMQDuplicateConsume getNoMQDuplicateConsumeAnnotation(ProceedingJoinPoint joinPoint) throws NoSuchMethodException {
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        Method targetMethod = joinPoint.getTarget().getClass().getDeclaredMethod(methodSignature.getName(), methodSignature.getMethod().getParameterTypes());
        return targetMethod.getAnnotation(NoMQDuplicateConsume.class);
    }

local key = KEYS[1] # 第一個(gè) Key，即冪等唯一標(biāo)識 uniqueKey
local value = ARGV[1] # 第一個(gè)參數(shù)，即初始化冪等消費(fèi)狀態(tài)，為消費(fèi)中
local expire_time_ms = ARGV[2] # 第二個(gè)參數(shù)，即冪等 Key 過期時(shí)間

return redis.call('SET', key, value, 'NX', 'GET', 'PX', expire_time_ms)

public class IdempotentConfiguration {
    /**
     * 防止消息隊(duì)列消費(fèi)者重復(fù)消費(fèi)消息切面控制器
     */
    @Bean
    public NoMQDuplicateConsumeAspect noMQDuplicateConsumeAspect(StringRedisTemplate stringRedisTemplate) {
        return new NoMQDuplicateConsumeAspect(stringRedisTemplate);
    }
}

public class SpELUtil {
    /**
     * 校驗(yàn)并返回實(shí)際使用的 spEL 表達(dá)式
     *
     * @param spEl spEL 表達(dá)式
     * @return 實(shí)際使用的 spEL 表達(dá)式
     */
    public static Object parseKey(String spEl, Method method, Object[] contextObj) {
        List<String> spELFlag = ListUtil.of("#", "T(");
        Optional<String> optional = spELFlag.stream().filter(spEl::contains).findFirst();
        if (optional.isPresent()) {
            return parse(spEl, method, contextObj);
        }
        return spEl;
    }

    /**
     * 轉(zhuǎn)換參數(shù)為字符串
     *
     * @param spEl       spEl 表達(dá)式
     * @param contextObj 上下文對象
     * @return 解析的字符串值
     */
    public static Object parse(String spEl, Method method, Object[] contextObj) {
        DefaultParameterNameDiscoverer discoverer = new DefaultParameterNameDiscoverer();
        ExpressionParser parser = new SpelExpressionParser();
        Expression exp = parser.parseExpression(spEl);
        String[] params = discoverer.getParameterNames(method);
        StandardEvaluationContext context = new StandardEvaluationContext();
        if (ArrayUtil.isNotEmpty(params)) {
            for (int len = 0; len < params.length; len++) {
                context.setVariable(params[len], contextObj[len]);
            }
        }
        return exp.getValue(context);
    }
}