支持redis分布式锁

pom.xml

@@ -243,6 +243,12 @@
         <version>2.9.0</version>
         <scope>provided</scope>
       </dependency>
+      <dependency>
+        <groupId>com.github.jedis-lock</groupId>
+        <artifactId>jedis-lock</artifactId>
+        <version>1.0.0</version>
+        <scope>provided</scope>
+      </dependency>
       <dependency>
         <groupId>org.redisson</groupId>
         <artifactId>redisson</artifactId>

weixin-java-common/pom.xml

@@ -118,6 +118,15 @@
       <artifactId>dom4j</artifactId>
       <version>2.1.1</version>
     </dependency>
+    <dependency>
+      <groupId>redis.clients</groupId>
+      <artifactId>jedis</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>com.github.jedis-lock</groupId>
+      <artifactId>jedis-lock</artifactId>
+    </dependency>
   </dependencies>
 
   <build>

weixin-java-common/src/main/java/me/chanjar/weixin/common/util/locks/JedisDistributedLock.java

@@ -0,0 +1,71 @@
+package me.chanjar.weixin.common.util.locks;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import redis.clients.jedis.Jedis;
+import redis.clients.util.Pool;
+
+/**
+ * JedisPool 分布式锁
+ *
+ * @author <a href="https://github.com/007gzs">007</a>
+ */
+public class JedisDistributedLock implements Lock {
+  private final Pool<Jedis> jedisPool;
+  private final JedisLock lock;
+
+  public JedisDistributedLock(Pool<Jedis> jedisPool, String key){
+    this.jedisPool = jedisPool;
+    this.lock = new JedisLock(key);
+  }
+
+  @Override
+  public void lock() {
+    try (Jedis jedis = jedisPool.getResource()) {
+      if (!lock.acquire(jedis)) {
+        throw new RuntimeException("acquire timeouted");
+      }
+    } catch (InterruptedException e) {
+      throw new RuntimeException("lock failed", e);
+    }
+  }
+
+  @Override
+  public void lockInterruptibly() throws InterruptedException {
+    try (Jedis jedis = jedisPool.getResource()) {
+      if (!lock.acquire(jedis)) {
+        throw new RuntimeException("acquire timeouted");
+      }
+    }
+  }
+
+  @Override
+  public boolean tryLock() {
+    try (Jedis jedis = jedisPool.getResource()) {
+      return lock.acquire(jedis);
+    } catch (InterruptedException e) {
+      throw new RuntimeException("lock failed", e);
+    }
+  }
+
+  @Override
+  public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
+    try (Jedis jedis = jedisPool.getResource()) {
+      return lock.acquire(jedis);
+    }
+  }
+
+  @Override
+  public void unlock() {
+    try (Jedis jedis = jedisPool.getResource()) {
+      lock.release(jedis);
+    }
+  }
+
+  @Override
+  public Condition newCondition() {
+    throw new RuntimeException("unsupported method");
+  }
+
+}

weixin-java-open/src/main/java/me/chanjar/weixin/open/api/WxOpenConfigStorage.java

@@ -6,6 +6,8 @@
 import me.chanjar.weixin.open.bean.WxOpenAuthorizerAccessToken;
 import me.chanjar.weixin.open.bean.WxOpenComponentAccessToken;
 
+import java.util.concurrent.locks.Lock;
+
 /**
  * @author <a href="https://github.com/007gzs">007</a>
  */
@@ -53,6 +55,10 @@ public interface WxOpenConfigStorage {
 
   WxMaConfig getWxMaConfig(String appId);
 
+  Lock getComponentAccessTokenLock();
+
+  Lock getLockByKey(String key);
+
   /**
    * 应该是线程安全的
    *

weixin-java-open/src/main/java/me/chanjar/weixin/open/api/impl/AbstractWxOpenInRedisConfigStorage.java

@@ -13,6 +13,9 @@ public abstract class AbstractWxOpenInRedisConfigStorage extends WxOpenInMemoryC
 
   protected final static String AUTHORIZER_REFRESH_TOKEN_KEY = "wechat_authorizer_refresh_token:";
   protected final static String AUTHORIZER_ACCESS_TOKEN_KEY = "wechat_authorizer_access_token:";
+
+  protected final static String LOCK_KEY = "wechat_lock:";
+
   protected final static String JSAPI_TICKET_KEY = "wechat_jsapi_ticket:";
   protected final static String CARD_API_TICKET_KEY = "wechat_card_api_ticket:";
 
@@ -26,6 +29,7 @@ public abstract class AbstractWxOpenInRedisConfigStorage extends WxOpenInMemoryC
   protected String authorizerAccessTokenKey;
   protected String jsapiTicketKey;
   protected String cardApiTicket;
+  protected String lockKey;
 
   @Override
   public void setComponentAppId(String componentAppId) {
@@ -36,8 +40,9 @@ public void setComponentAppId(String componentAppId) {
     componentAccessTokenKey = prefix + COMPONENT_ACCESS_TOKEN_KEY.concat(componentAppId);
     authorizerRefreshTokenKey = prefix + AUTHORIZER_REFRESH_TOKEN_KEY.concat(componentAppId);
     authorizerAccessTokenKey = prefix + AUTHORIZER_ACCESS_TOKEN_KEY.concat(componentAppId);
-    this.jsapiTicketKey = JSAPI_TICKET_KEY.concat(componentAppId);
-    this.cardApiTicket = CARD_API_TICKET_KEY.concat(componentAppId);
+    lockKey = prefix + LOCK_KEY.concat(componentAppId);
+    jsapiTicketKey = prefix + JSAPI_TICKET_KEY.concat(componentAppId);
+    cardApiTicket = prefix + CARD_API_TICKET_KEY.concat(componentAppId);
   }
 
   protected String getKey(String prefix, String appId) {

weixin-java-open/src/main/java/me/chanjar/weixin/open/api/impl/WxOpenComponentServiceImpl.java

@@ -23,6 +23,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.locks.Lock;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
@@ -109,18 +110,28 @@ public boolean checkSignature(String timestamp, String nonce, String signature)
 
   @Override
   public String getComponentAccessToken(boolean forceRefresh) throws WxErrorException {
-
-    if (this.getWxOpenConfigStorage().isComponentAccessTokenExpired() || forceRefresh) {
+    final WxOpenConfigStorage config = this.getWxOpenConfigStorage();
+    if (!config.isComponentAccessTokenExpired() && !forceRefresh) {
+      return config.getComponentAccessToken();
+    }
+    Lock lock = config.getComponentAccessTokenLock();
+    lock.lock();
+    try {
+      if (!config.isComponentAccessTokenExpired() && !forceRefresh) {
+        return config.getComponentAccessToken();
+      }
       JsonObject jsonObject = new JsonObject();
       jsonObject.addProperty("component_appid", getWxOpenConfigStorage().getComponentAppId());
       jsonObject.addProperty("component_appsecret", getWxOpenConfigStorage().getComponentAppSecret());
       jsonObject.addProperty("component_verify_ticket", getWxOpenConfigStorage().getComponentVerifyTicket());
 
       String responseContent = this.getWxOpenService().post(API_COMPONENT_TOKEN_URL, jsonObject.toString());
       WxOpenComponentAccessToken componentAccessToken = WxOpenComponentAccessToken.fromJson(responseContent);
-      getWxOpenConfigStorage().updateComponentAccessToken(componentAccessToken);
+      config.updateComponentAccessToken(componentAccessToken);
+      return config.getComponentAccessToken();
+    } finally {
+      lock.unlock();
     }
-    return this.getWxOpenConfigStorage().getComponentAccessToken();
   }
 
   @Override
@@ -346,18 +357,28 @@ public void setAuthorizerOption(String authorizerAppid, String optionName, Strin
 
   @Override
   public String getAuthorizerAccessToken(String appId, boolean forceRefresh) throws WxErrorException {
-
-    if (this.getWxOpenConfigStorage().isAuthorizerAccessTokenExpired(appId) || forceRefresh) {
+    WxOpenConfigStorage config = getWxOpenConfigStorage();
+    if (!config.isAuthorizerAccessTokenExpired(appId) && !forceRefresh){
+      return config.getAuthorizerAccessToken(appId);
+    }
+    Lock lock = config.getWxMpConfigStorage(appId).getAccessTokenLock();
+    lock.lock();
+    try{
+      if (!config.isAuthorizerAccessTokenExpired(appId) && !forceRefresh){
+        return config.getAuthorizerAccessToken(appId);
+      }
       JsonObject jsonObject = new JsonObject();
       jsonObject.addProperty("component_appid", getWxOpenConfigStorage().getComponentAppId());
       jsonObject.addProperty("authorizer_appid", appId);
       jsonObject.addProperty("authorizer_refresh_token", getWxOpenConfigStorage().getAuthorizerRefreshToken(appId));
       String responseContent = post(API_AUTHORIZER_TOKEN_URL, jsonObject.toString());
 
       WxOpenAuthorizerAccessToken wxOpenAuthorizerAccessToken = WxOpenAuthorizerAccessToken.fromJson(responseContent);
-      getWxOpenConfigStorage().updateAuthorizerAccessToken(appId, wxOpenAuthorizerAccessToken);
+      config.updateAuthorizerAccessToken(appId, wxOpenAuthorizerAccessToken);
+      return config.getAuthorizerAccessToken(appId);
+    }finally {
+      lock.unlock();
     }
-    return this.getWxOpenConfigStorage().getAuthorizerAccessToken(appId);
   }
 
   @Override

weixin-java-open/src/main/java/me/chanjar/weixin/open/api/impl/WxOpenInMemoryConfigStorage.java

@@ -44,6 +44,9 @@ public class WxOpenInMemoryConfigStorage implements WxOpenConfigStorage {
   private Map<String, Token> authorizerAccessTokens = new ConcurrentHashMap<>();
   private Map<String, Token> jsapiTickets = new ConcurrentHashMap<>();
   private Map<String, Token> cardApiTickets = new ConcurrentHashMap<>();
+  private Map<String, Lock> locks = new ConcurrentHashMap<>();
+
+  private Lock componentAccessTokenLock = getLockByKey("componentAccessTokenLock");
 
 
   @Override
@@ -61,6 +64,21 @@ public void updateComponentAccessToken(WxOpenComponentAccessToken componentAcces
     updateComponentAccessToken(componentAccessToken.getComponentAccessToken(), componentAccessToken.getExpiresIn());
   }
 
+  @Override
+  public Lock getLockByKey(String key){
+    Lock lock = locks.get(key);
+    if (lock == null) {
+      synchronized (WxOpenInMemoryConfigStorage.class){
+        lock = locks.get(key);
+        if (lock == null) {
+          lock = new ReentrantLock();
+          locks.put(key, lock);
+        }
+      }
+    }
+    return lock;
+  }
+
   @Override
   public WxMpConfigStorage getWxMpConfigStorage(String appId) {
     return new WxOpenInnerConfigStorage(this, appId);
@@ -211,13 +229,16 @@ private static class WxOpenInnerConfigStorage implements WxMpConfigStorage, WxMa
      * 云环境ID
      */
     private volatile String cloudEnv;
-    private Lock accessTokenLock = new ReentrantLock();
-    private Lock jsapiTicketLock = new ReentrantLock();
-    private Lock cardApiTicketLock = new ReentrantLock();
+    private final Lock accessTokenLock;
+    private final Lock jsapiTicketLock;
+    private final Lock cardApiTicketLock;
 
     private WxOpenInnerConfigStorage(WxOpenConfigStorage wxOpenConfigStorage, String appId) {
       this.wxOpenConfigStorage = wxOpenConfigStorage;
       this.appId = appId;
+      this.accessTokenLock = wxOpenConfigStorage.getLockByKey(appId + ":accessTokenLock");
+      this.jsapiTicketLock = wxOpenConfigStorage.getLockByKey(appId + ":jsapiTicketLock");
+      this.cardApiTicketLock = wxOpenConfigStorage.getLockByKey(appId + ":cardApiTicketLock");
     }
 
     @Override

weixin-java-open/src/main/java/me/chanjar/weixin/open/api/impl/WxOpenInRedisConfigStorage.java

@@ -1,9 +1,12 @@
 package me.chanjar.weixin.open.api.impl;
 
+import me.chanjar.weixin.common.util.locks.JedisDistributedLock;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisPool;
 import redis.clients.util.Pool;
 
+import java.util.concurrent.locks.Lock;
+
 /**
  * @author <a href="https://github.com/007gzs">007</a>
  */
@@ -163,4 +166,9 @@ public void updateCardApiTicket(String appId, String cardApiTicket, int expiresI
       jedis.setex(this.getKey(this.cardApiTicket, appId), expiresInSeconds - 200, cardApiTicket);
     }
   }
+
+  @Override
+  public Lock getLockByKey(String key) {
+    return new JedisDistributedLock(jedisPool, getKey(lockKey, key));
+  }
 }