分布式鎖獲取token

package com.sankuai.qcs.regulation.nanjing.util;

import com.dianping.squirrel.client.StoreKey;
import com.dianping.squirrel.client.impl.redis.RedisStoreClient;
import com.dianping.zebra.util.StringUtils;
import com.google.common.collect.Maps;
import com.sankuai.meituan.config.MtConfigClient;
import com.sankuai.qcs.regulation.nanjing.Conf;
import org.apache.http.HttpException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

/**
 * Describe:
 * Created by tanxiaolei
 * Date: 2018/4/20 11:50
 */
@Component
public class TokenUtils {

    private static final Logger LOGGER = LoggerFactory.getLogger(TokenUtils.class);

    @Resource
    private RedisStoreClient redisStoreClient;
    @Resource
    private MtConfigClient mtConfigClient;

    private static final String KEY_CATEGORY = "regulation_traffic";
    private static final String TOKEN_KEY_PARAMS = "nanjing_token_key";
    //緩存失效時間 11個小時
    private static final int TOKEN_EXPIRE_SECONDS = 39600;

    private static final String LOCK_KEY_PARAMS = "nanjing_lock_key";
    //分布式鎖失效時間2秒
    private static final int LOCK_EXPIRE_SECONDS = 2;

    private static final String NJ_TOKEN_USERID = "NJ_TOKEN_USERID";

    private static final Map<String, String> headers = Maps.newHashMap();

    static {
        headers.put("Connection", "keep-alive");
        headers.put("Accept-Charset", Conf.DEFAULT_CHARSET);
        headers.put("Content-Type", Conf.ContentType.JSON.getMimeType());
    }

    /**
     * 判斷token是否在redis存在
     *
     * @return
     */
    public boolean tokenExists() {
        StoreKey key = new StoreKey(KEY_CATEGORY, TOKEN_KEY_PARAMS);
        return redisStoreClient.exists(key);
    }

    /**
     * 刪除指定token
     *
     * @return
     */
    public void delToken() {
        StoreKey key = new StoreKey(KEY_CATEGORY, TOKEN_KEY_PARAMS);
        LOGGER.info("key : {} delete {}", key, redisStoreClient.delete(key));
    }

    /**
     * 獲取token
     *
     * @return
     */
    public String getToken() {
        StoreKey key = new StoreKey(KEY_CATEGORY, TOKEN_KEY_PARAMS);
        String token = redisStoreClient.get(key);
        LOGGER.info("get token ：{}  from redis", token);
        if (token == null) {
            StoreKey lock = new StoreKey(KEY_CATEGORY, LOCK_KEY_PARAMS);
            //分布式鎖，如果沒拿到鎖，則直接放棄，防止南京側服務出現問題，影響MQ消費
            if (redisStoreClient.setnx(lock, "lock", LOCK_EXPIRE_SECONDS)) {
                //雙重檢驗，防止重復獲取token
                token = redisStoreClient.get(key);
                if (token == null) {
                    try {
                        String userId = mtConfigClient.getValue(NJ_TOKEN_USERID);
                        LOGGER.info("mtConfigClient get userId : {}", userId);
                        token = HttpClientUtils.post("http://" + Conf.GET_TOKEN_URL + userId, "320100", headers);
                        LOGGER.info("get token : {} from http", token);
                        if (StringUtils.isNotBlank(token)) {
                            redisStoreClient.set(key, token, TOKEN_EXPIRE_SECONDS);
                        }
                    } catch (HttpException e) {
                        LOGGER.error("get token errer", e);
                    }
                }
                //將分布式鎖直接過期
                redisStoreClient.expire(lock, 0);
            }
        }
        return token;
    }
}

package com.sankuai.qcs.regulation.nanjing.util;

import com.sankuai.qcs.regulation.nanjing.Conf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.websocket.ClientEndpoint;
import javax.websocket.CloseReason;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;

/**
 * Describe:
 * Created by tanxiaolei
 * Date: 2018/4/18 14:26
 */
@ClientEndpoint
@Component
public class WebSocketClientUtils {


//    @Autowired
//    private TokenUtils tokenUtils;

    private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketClientUtils.class);

    private static ApplicationContext applicationContext;

    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    public static void setApplicationContext(ApplicationContext applicationContext) {
        WebSocketClientUtils.applicationContext = applicationContext;
    }

    @OnOpen
    public void onOpen(Session session) {
        //經過試驗，客戶端設置 buffer size時並不生效
        session.setMaxBinaryMessageBufferSize(Conf.BINARY_MESSAGE_BUFFER_SIZE);
        session.setMaxTextMessageBufferSize(Conf.BINARY_MESSAGE_BUFFER_SIZE);
        LOGGER.info("Session {}, {} Connected", session.getId(), session.getRequestParameterMap());
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        LOGGER.info("Session receive message : {}", message);
        //如果是403，表示token失效
        if ("403".equals(message)) {
            delAndGetNewToken();
        }
    }

    @OnClose
    public void onClose(Session session, CloseReason closeReason) {
        LOGGER.info("Session max buffer size {} {} close because of {}", session.getMaxBinaryMessageBufferSize(), session.getRequestParameterMap(), closeReason);
    }

    @OnError
    public void onError(Session session, Throwable throwable) {
        if (session != null) {
            LOGGER.error("Session {} error", session.getRequestParameterMap(), throwable);
        } else {
            LOGGER.error("error", throwable);
        }
    }

    private void delAndGetNewToken() {
        TokenUtils tokenUtils = (TokenUtils) applicationContext.getBean(TokenUtils.class);
        LOGGER.info("toeknUtils : {}", tokenUtils);
        tokenUtils.delToken();
        LOGGER.info("again get token : {}", tokenUtils.getToken());
    }

}

 

    /**
     * 添加 Key 對應的值為 Value，只有當 Key 不存在時才添加，如果 Key 已經存在，不改變現有的值
     * {@link RedisStoreClient#add(StoreKey, Object, int)}
     * @param key 要添加的  Key
     * @param value 要添加的 Value
     * @param expireInSeconds 過期時間
     * @return 如果 Key 不存在且添加成功，返回 true<br>
     *         如果 Key 已經存在，返回 false
     * @throws StoreException 異常都是 StoreException 的子類且是 RuntimeException，可以根據需要捕獲相應異常。
     *         如：如果需要捕獲超時異常，可以捕獲 StoreTimeoutException
     */
    public Boolean setnx(StoreKey key, Object value, int expireInSeconds);

 問題的關鍵是：方法：

getToken
使用了加鎖方法：

if (redisStoreClient.setnx(lock, "lock", LOCK_EXPIRE_SECONDS)) {
這個方法 如果 Key 不存在且添加成功， 如果 Key 已經存在，返回 false
也就是說：只有key添加成功的話才獲取token，否則丟棄，防止南京服務器出問題；