java - 在多个java线程之间共享数据并获取更新值

Question

我想创建一个 java 应用程序，我们想在访问 token 的帮助下为多个用户进行休息调用。我每个用户使用 1 个线程。我正在使用的访问 token 有效期为 1 小时。 token 过期后，我将收到 401 错误，并且必须为所有线程更新 token ，然后继续。我正在考虑使用我已设为静态的 volatile 变量来更新所有线程。我的要求是，当我在其中一个线程中知道 token 已过期时，我希望所有线程停止处理并等待新 token 生成(这需要几秒钟)。此外，一旦生成， token 应自动更新，每个线程不会因 token 过期而失败。

下面是我写的示例代码:

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Sample {

public static void main(String[] args) {

    String[] myStrings = { "User1" , "User2" , "User3" };

    ScheduledExecutorService scheduledExecutorService = Executors
            .newScheduledThreadPool(myStrings.length);

    TokenGenerator.getToken();

    for(String str : myStrings){
        scheduledExecutorService.scheduleAtFixedRate(new Task(str), 0, 5, TimeUnit.SECONDS);    
    }
}

}

 class Task implements Runnable{

private String name;

public Task(String name){
    this.name = name;

}


@Override
public void run() {

    getResponse(TokenGenerator.token);

}

private void getResponse(String token) {
    // Make http calls
    // if token expire , call getToken again. Pause all the running threads , and
    // update the token for all threads

    TokenGenerator.getToken();
}

 }

 class TokenGenerator {

public static volatile String token;

public static void getToken() {

    token = "new Token everytime";

}

 }

有没有更好的方法来解决这个问题？上面的代码不满足我的用例，因为一旦线程开始生成新 token ，所有其他线程都不会被暂停。请求提出一些改进建议..

Answer 1

您可以将 token 放在 AtomicReference 中并使用 Semaphore暂停线程:

public class TokenWrapper {
  private final AtomicReference<Token> tokenRef = new AtomicReference<>(null);
  private final Semaphore semaphore = new Semaphore(Integer.MAX_VALUE);

  public TokenWrapper() {
    Token newToken = // refresh token
    tokenRef.set(newToken);
  }

  public Token getToken() {
    Token token = null;
    while((token = tokenRef.get()) == null) {
      semaphore.acquire();
    }
    return token;
  }

  public Token refreshToken(Token oldToken) {
    if(tokenRef.compareAndSet(oldToken, null)) {
      semaphore.drainPermits();          
      Token newToken = // refresh token
      tokenRef.set(newToken);
      semaphore.release(Integer.MAX_VALUE);
      return newToken;
    } else return getToken();
  }
}

public class RESTService {
  private static final TokenWrapper tokenWrapper = new TokenWrapper();

  public void run() {
    Token token = tokenWrapper.getToken();
    Response response = // call service with token
    if(response.getStatus == 401) {
      tokenWrapper.refreshToken(token);
    }
  }
}

refreshToken()使用原子 compareAndSet在 tokenRef确保只有一个线程会刷新 token ，然后调用 drainPermits()在 semaphore 上使其他线程等待直到 token 被刷新。 getToken()如果不是 null，则返回 token , 否则等待 semaphore - 这是在循环中完成的，因为线程可能必须在 tokenRef 之间旋转几个周期。被设置为 null和 drainPermits()被拜访semaphore .

---

编辑:修改了refreshToken(Token oldToken)的签名以便传递旧 token 而不是在方法内部读取 - 这是为了防止 RESTService_A 刷新 token ，RESTService_B 获取带有旧过期 token 的 401，然后 RESTService_B 调用 refreshToken 的情况。在 RESTService_A 调用 refreshToken 之后已完成，导致 token 被刷新两次。使用新签名，RESTService_B 将传入旧的过期 token ，因此 compareAndSet当旧 token 无法匹配新 token 时调用将失败，导致 refreshToken只被调用一次。