基于Mutiny的Token缓存实现线程安全问题咨询

问题描述 #

我是响应式编程新手，正学习基础概念。我想实现如下逻辑：若认证Token仍有效则返回，否则请求新Token。我编写的代码如下：

public class AsyncTokenSupplier implements Supplier<Uni<Token>> {
   private Token cachedToken;

   @Override
   public Uni<Token> get() {
        if (isTokenValid(cachedToken)) {
            return Uni.createFrom().item(cachedToken);
        } else {
            return doGetToken().onItem().invoke(token -> {
                cachedToken = token;
            });
        }
    }

   private Uni<Token> doGetToken() {
      //Api call here
   }

   private boolean isTokenValid(Token token) {
      if (token == null) {
         return false;
      }
      //Expiration check here
   }
}

但该代码存在线程安全问题：当多个线程并行执行get()方法时，可能会同时发起多次新Token请求，希望得到相关指导与建议。

解决方案 #

1. 原子引用+请求防重复方案 #

用AtomicReference存储缓存Token和正在进行的请求，通过CAS操作确保同一时间只有一个请求发起Token刷新，其他请求等待同一个异步结果：

import java.util.concurrent.atomic.AtomicReference;
import io.smallrye.mutiny.Uni;

public class AsyncTokenSupplier implements Supplier<Uni<Token>> {
    private final AtomicReference<Token> cachedToken = new AtomicReference<>();
    private final AtomicReference<Uni<Token>> pendingRequest = new AtomicReference<>();

    @Override
    public Uni<Token> get() {
        Token current = cachedToken.get();
        if (isTokenValid(current)) {
            return Uni.createFrom().item(current);
        }

        // 检查是否已有正在进行的请求
        Uni<Token> existingRequest = pendingRequest.get();
        if (existingRequest != null) {
            return existingRequest;
        }

        // 创建新的Token请求
        Uni<Token> newRequest = doGetToken()
                .onItem().invoke(token -> {
                    cachedToken.set(token);
                    pendingRequest.set(null); // 请求完成后清空标记
                })
                .onFailure().invoke(throwable -> {
                    pendingRequest.set(null); // 失败也要清空，避免阻塞后续请求
                });

        // CAS操作确保只有第一个线程能设置成功
        if (pendingRequest.compareAndSet(null, newRequest)) {
            return newRequest;
        } else {
            return pendingRequest.get();
        }
    }

    private Uni<Token> doGetToken() {
        // 替换为实际的认证API调用逻辑
        return Uni.createFrom().item(new Token());
    }

    private boolean isTokenValid(Token token) {
        if (token == null) return false;
        // 替换为实际的过期时间校验逻辑
        return token.getExpiresAt() > System.currentTimeMillis();
    }
}

// 示例Token类
class Token {
    private long expiresAt;

    public long getExpiresAt() {
        return expiresAt;
    }

    public void setExpiresAt(long expiresAt) {
        this.expiresAt = expiresAt;
    }
}

2. 利用Mutiny的Memoize特性（简洁版） #

Mutiny内置的memoize操作符可以自动合并重复请求，并根据Token有效期自动刷新，无需手动处理原子引用：

import io.smallrye.mutiny.Uni;

public class AsyncTokenSupplier implements Supplier<Uni<Token>> {
    private Uni<Token> cachedTokenUni;

    @Override
    public Uni<Token> get() {
        if (cachedTokenUni == null || !isCachedTokenValid()) {
            cachedTokenUni = doGetToken()
                    // 缓存直到Token失效，失效后自动重新请求
                    .memoize().until(this::isTokenValid);
        }
        return cachedTokenUni;
    }

    private boolean isCachedTokenValid() {
        try {
            Token token = cachedTokenUni.await().indefinitely();
            return isTokenValid(token);
        } catch (Exception e) {
            return false;
        }
    }

    private Uni<Token> doGetToken() {
        // 替换为实际的认证API调用逻辑
        return Uni.createFrom().item(new Token());
    }

    private boolean isTokenValid(Token token) {
        if (token == null) return false;
        return token.getExpiresAt() > System.currentTimeMillis();
    }
}

核心注意事项 #

• 避免竞态：两种方案都确保了同一时间只有一个Token刷新请求发起，防止重复调用认证API。
• 故障恢复：请求失败后必须清空等待标记或让缓存失效，避免后续请求一直阻塞在失败结果上。
• 响应式语义：尽量使用Mutiny提供的操作符处理异步逻辑，避免在响应式代码中使用不必要的阻塞操作。

内容的提问来源于stack exchange，提问作者Evgeny