* 三方平台支持redis分布式锁;getComponentAccessToken 加锁 * getAuthorizerAccessToken 加锁dev1
@@ -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> | |||
@@ -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> | |||
@@ -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"); | |||
} | |||
} |
@@ -6,6 +6,8 @@ import me.chanjar.weixin.mp.config.WxMpConfigStorage; | |||
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); | |||
/** | |||
* 应该是线程安全的 | |||
* | |||
@@ -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 abstract class AbstractWxOpenInRedisConfigStorage extends WxOpenInMemoryC | |||
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) { | |||
@@ -23,6 +23,7 @@ import org.slf4j.LoggerFactory; | |||
import java.util.List; | |||
import java.util.Map; | |||
import java.util.concurrent.locks.Lock; | |||
import java.util.concurrent.ConcurrentHashMap; | |||
/** | |||
@@ -109,8 +110,16 @@ public class WxOpenComponentServiceImpl implements WxOpenComponentService { | |||
@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()); | |||
@@ -118,9 +127,11 @@ public class WxOpenComponentServiceImpl implements WxOpenComponentService { | |||
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,8 +357,16 @@ public class WxOpenComponentServiceImpl implements WxOpenComponentService { | |||
@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); | |||
@@ -355,9 +374,11 @@ public class WxOpenComponentServiceImpl implements WxOpenComponentService { | |||
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 | |||
@@ -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 class WxOpenInMemoryConfigStorage implements WxOpenConfigStorage { | |||
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 @@ public class WxOpenInMemoryConfigStorage implements WxOpenConfigStorage { | |||
* 云环境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 | |||
@@ -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 class WxOpenInRedisConfigStorage extends AbstractWxOpenInRedisConfigStora | |||
jedis.setex(this.getKey(this.cardApiTicket, appId), expiresInSeconds - 200, cardApiTicket); | |||
} | |||
} | |||
@Override | |||
public Lock getLockByKey(String key) { | |||
return new JedisDistributedLock(jedisPool, getKey(lockKey, key)); | |||
} | |||
} |