关于亿级流量网站架构一书缓存机制的探讨

Posted 2020-10-07 lin_sen

在京东的亿级流量网站架构一书，175页介绍缓存有这样一段话

仅就这段代码来看，在高并发情况下，实际上并不能阻止大量线程调用loadSync函数

当然这个书里的代码是作者的简写，这里探讨只是针对书中这段代码，实际生成代码应该有考虑这个问题，另外loadSync函数的逻辑看不到，也可能有考虑到到这个问题。

这中情况应该使用双锁，另外firstCreateNewEntry也应该是定义为volatile类型。还有如果是分布式缓存，针对远程客户端的回源请求应该要设置一个时限，比如30秒内只受理一个回源请求。

下面给出本人项目中使用的缓存加载机制。本人项目根据机构，应用，数据库类型三个字段进行了分库。因此缓存最粗粒度也是这个级别的。

高并发治理关键点：

• 初始获取锁对象时使用双锁机制。
• 使用获取到的锁对象同步代码。
• 记录上次重刷时间的 lastReloadDate来自ConcurrentHashMap对象，对象在jvm的堆中，所以无需volatile类型就能保证线程间的可见性。
• 最终加载数据时没有使用双锁，因为本项目是使用分布式缓存，都是由远程客户端发起的回源请求，双锁只能保证本地缓存高并发时刻多线程不会同时进入，而不能防止远程回源。因为远程调用时陆陆续续到达的，这里假设30秒缓存能加载完成，一旦加载完成就不会有客户端要求回源。该机制保障了30秒内的回源请求只会触发一次缓存加载。（客户端发现无缓存，发起回源请求，由于网络延迟，请求还未到达缓存加载服务器，这时即使缓存已经加载完成了，如果不防范，这些在路上的回源请求也会被受理）

/**
 * 作者: 林森
 * 日期: 2017年1月5日    
 * CopyRight @lins    
 */
package com.yunkang.ykcachemanage.provider.service;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.ColumnMapRowMapper;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import com.alibaba.dubbo.config.annotation.Service;
import com.yunkang.ykcachemanage.keys.CacheKeys.cache_keys;
import com.yunkang.ykcachemanage.keys.CacheKeys;
import com.yunkang.ykcachemanage.keys.DictCacheLoader;
import com.yunkang.yktechcom.cache.DictCacheHelper;
import com.yunkang.yktechcom.jdbc.JDBCRouteHelper;

/**
 * 
 * 项目名称：ykcachemanage-provider
 * 
 * 类描述：
 * 
 * 从数据库获取数据并缓存到redis中 为防止多个线程并发重刷redis可能导致无限循环加载问题,该服务建议只开一一个,同时刷新方法提供同步保护
 * 另外限制同一个字典十秒内只能刷新一次
 * 该服务一般用于重置或初始化redis,初始化后redis与mysql的同步由业务系统的在维护字典时,手工调用缓存更新api保证.
 * 
 * 
 * 创建人：林森
 * 
 * 创建时间：2017年1月5日 修改人： 修改时间： 修改备注：
 * 
 * @version
 * 
 */
@Service
public class DictCacheLoaderProvider implements DictCacheLoader {

    @Autowired
    JDBCRouteHelper jdbcRouteHelper;
    @Autowired
    DictCacheHelper dictCacheHelper;

    //针对不同的缓存集合，使用不同的锁
    Map<String, Object> mapLock = new ConcurrentHashMap<>();

    //记录上次重刷缓存的时间，用于防止恶意重刷。
    static Map<String, Date> lastReload = new ConcurrentHashMap<>();

    @Override
    public void reloadCache(String idFieldName, cache_keys setKey, Map<String, Object> splitFields,
            Map<String, Object> filter, Map<String, String> dbParam) {
        String[] splitFieldValue = splitFields.values().toArray(new String[0]);
        String lockKey = CacheKeys.getPrefix(splitFieldValue) + setKey;

        if (mapLock.get(lockKey) == null) {
            synchronized (this) {
                if (mapLock.get(lockKey) == null) {
                    mapLock.put(lockKey, new Object());
                }
            }
        }
        Object lock = mapLock.get(lockKey);
        synchronized (lock) {
            Date lastReloadDate = lastReload.get(lockKey);
            if (lastReload.get(lockKey) == null || (new Date().getTime() - lastReloadDate.getTime()) > 30 * 1000) {// 30秒内不重刷
                dictCacheHelper.removeAll(CacheKeys.getPrefix(splitFieldValue) + setKey);
                this.reloadCacheFromDB(idFieldName, setKey, splitFields, filter,dbParam);
                lastReload.put(lockKey, new Date());
            } else {
                // ===rejectreload redis from db " + setKey + Thread.currentThread());
            }
        }
    }

    static String sqlGetDoctor = "select * from doctors where status=\'1\'";
    static String sqlDefault = "select * from %s where 1=1 ";


    private void reloadCacheFromDB(String idFieldName, cache_keys setKey, Map<String, Object> splitFields,
            Map<String, Object> filter, Map<String, String> dbParam) {
        switch (setKey) {
        case KHLIS_DOCTORS:
            doLoadFromDB(setKey, idFieldName, splitFields, sqlGetDoctor, filter,  dbParam);
            break;
        default:
            doLoadFromDB(setKey, idFieldName, splitFields, getSqlFromSetKey(setKey, sqlDefault), filter, dbParam);
            break;
        }
    }

    // 使用setkey推测数据表的名字.
    private String getSqlFromSetKey(cache_keys setKey, String sql) {
        String tableName = setKey.toString().substring(setKey.toString().indexOf(\'_\') + 1).toLowerCase();
        return String.format(sql, tableName);
    }

    /**
     * @param setKey 用于确定要加载的字典表
     * @param idFieldName 字典表的主键名称,多个冒号隔开
     * @param splitFields 字典表切割字段,将字典表切割为多个缓存.
     * @param filter 只有满足条件的才会加载到缓存
     * @param dbParam 对应数据路由表的用于分库的三个字段 org_id,app_id,dbs_type
     *    从数据库读取字典表的数据后更新缓存    
     */
    private void doLoadFromDB(cache_keys setKey, String idFieldName, Map<String, Object> splitFields, String sql,
            Map<String, Object> filter, Map<String, String> dbParam) {
        String orgId =  dbParam.get("org_id");
        String appId =  dbParam.get("app_id");
        String dbsType =  dbParam.get("dbs_type");
        if (orgId == null || appId == null|| dbsType == null)
            throw new RuntimeException("org_id,app_id,dbs_type必须有,否则无法加载缓存");
 
        String[] splitFieldValues = splitFields.values().toArray(new String[0]);

        List<Map<String, Object>> po = new ArrayList<>();
        StringBuilder sb = new StringBuilder();
        sb.append(sql);
        for (Entry<String, Object> item : splitFields.entrySet()) {
            sb.append(String.format(" and %s=:%s ", item.getKey(), item.getKey()));
        }
        Map<String, Object> paramMap = new HashMap<String, Object>();

        if (filter != null) {
            for (Entry<String, Object> item : filter.entrySet()) {
                sb.append(String.format(" and %s=:%s ", item.getKey(), item.getKey()));
            }
            paramMap.putAll(filter);
        }
        paramMap.putAll(splitFields);
        MapSqlParameterSource paramSource = new MapSqlParameterSource(paramMap);
        NamedParameterJdbcTemplate jdbc = jdbcRouteHelper.getJDBCTemplate(orgId, appId,dbsType);
        TransactionTemplate trans = jdbcRouteHelper.getTransactionTemplate(orgId, appId,dbsType);
        po = (List<Map<String, Object>>) trans
                .execute(new TransactionCallback<List<Map<String, Object>>>() {
                    public List<Map<String, Object>> doInTransaction(TransactionStatus status) {
                        return jdbc.query(sb.toString(), paramSource,
                                new ColumnMapRowMapper());
                    }
                });

        Map<String, Object> mapValues = new HashMap<>();
        for (Map<String, Object> item : po) {
            if (idFieldName.contains(":")) {// 多字段主键
                String[] ids = idFieldName.split(":");
                String keys = "";
                for (String _id : ids) {
                    keys += item.get(_id).toString() + ":";
                }
                mapValues.put(keys, item);
            } else {
                mapValues.put(item.get(idFieldName).toString(), item);
            }

        }
        dictCacheHelper.setAll(CacheKeys.getPrefix(splitFieldValues) + setKey, mapValues);

    }

}