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

Question

我想创建一个java应用程序,我们希望在访问令牌的帮助下为多个用户进行休息调用.我每个用户使用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,然后RESTService_B调用refreshToken后RESTService_A的调用refreshToken已完成,导致令牌刷新两次.使用新签名时,RESTService_B将传入旧的过期令牌,因此compareAndSet当旧令牌无法匹配新令牌时,调用将失败,从而导致refreshToken仅被调用一次.