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

Question

我想创建一个Java应用程序，在这里我们要借助访问令牌对多个用户进行rest调用。我正在为每个用户使用1个线程。我正在使用的访问令牌有效期为1个小时。一旦令牌过期，我将收到401错误，并且必须为所有线程更新令牌，然后继续。我正在考虑使用我已将其设为静态的volatile变量来更新所有线程。我的要求是，在我知道其中一个线程令牌已过期的那一刻，我希望所有线程停止处理并等待直到生成新令牌（这需要几秒钟）。令牌应该自动更新，而每个线程都不会由于令牌过期而失败。

以下是我编写的示例代码：

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";

}

 }

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

Answer 1

您可以将令牌放入AtomicReference中，并使用信号量来暂停线程：

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，以确保只有一个线程将刷新令牌，然后调用drainPermits()上semaphore导致其他线程等待，直到令牌被刷新。
getToken()如果不是null，则返回令牌，否则等待semaphore-循环完成，因为线程可能必须在tokenRef设置为null和drainPermits()被调用之间旋转几个周期semaphore。

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