商品中心—缓存与 DB 一致性的技术文档

CREATE TABLE `data_refresh_detail` (    `id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',    `cache_key` varchar(128) NOT NULL DEFAULT '' COMMENT '缓存key',    `operation_type` tinyint(1) NOT NULL DEFAULT '1' COMMENT '1新增/修改，2删除',    `cache_json` text NOT NULL COMMENT '缓存内容',    `cache_type` tinyint(1) DEFAULT NULL COMMENT '缓存类型，1Redis，2Tair',    `cache_status` tinyint(1) DEFAULT '0' COMMENT '缓存的处理状态，默认为0未处理，1为已处理',    `version` varchar(32) NOT NULL COMMENT '消息版本号',    `message_type` tinyint(1) NOT NULL COMMENT '是否热点消息，0普通，1热点',    `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '创建时间',    `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',    PRIMARY KEY (`id`) USING BTREE) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8 COMMENT='缓存消息DB记录表';

@Retention(RetentionPolicy.RUNTIME)@Target(ElementType.METHOD)public @interface CacheRefresh {    //缓存消息key    String mqCacheKey();    //缓存的key    String cacheKey();    //需要缓存的值在方法参数中的坐标偏移量，默认不传取第一个参数    String index() default "0";    //是否热点缓存消息：0普通缓存消息，1热点缓存消息    String messageType() default "0";    //缓存的操作类型：1新增/修改，2删除    String operationType() default "1";    //缓存的组件类别：1是Redis，2是Tair    String cacheType() default "1";}

@Component@Datapublic class InventoryBucketCache {    ...    //本地存储关于分桶信息    @CacheRefresh(cacheKey = "bucketKey", mqCacheKey = CacheConstant.INVENTORY_SKU_KEY, index = "1", messageType = CacheConstant.MESSAGE_TYPE_HOT, cacheType = CacheConstant.TAIR_CACHE_TYPE)    public void setBucketLocalCache(String bucketKey, BucketLocalCache bucketLocalCache) {        ...    }    ...}

//刷新缓存的自定义注解@Aspect@Componentpublic class CacheRefreshAspect {    @Autowired    private DataRefreshProducer producer;        @Autowired    private CacheRefreshConverter cacheRefreshConverter;        @Autowired    private CacheQueue cacheQueue;
    //切入点，@CacheRefresh注解标注的    @Pointcut("@annotation(com.demo.eshop.cache.annotation.CacheRefresh)")        public void pointcut() {    }
    //环绕通知，在方法执行前后    //@param point 切入点    //@return 结果    @Around("pointcut() && @annotation(cacheRefresh)")    public Object around(ProceedingJoinPoint point, CacheRefresh cacheRefresh) throws Throwable {        //签名信息        Signature signature = point.getSignature();        //强转为方法信息        MethodSignature methodSignature = (MethodSignature) signature;        //参数名称        String[] parameterNames = methodSignature.getParameterNames();        //参数值        Object[] parameterValues = point.getArgs();        Object response;        try {            //先执行本地方法再执行异步的操作            response = point.proceed();        } catch (Throwable throwable) {            log.error("执行方法: {}失败，异常信息: {}", methodSignature.getMethod().getName(), throwable);            throw throwable;        }          try {            MessageCache messageCache = new MessageCache();            for (int i = 0; i < parameterValues.length; i++) {                if (parameterNames[i].equals(cacheRefresh.cacheKey())) {                    messageCache.setCacheKey(String.valueOf(parameterValues[i]));                }                if (Integer.valueOf(cacheRefresh.index()) == i) {                    messageCache.setCacheJSON(JSONObject.toJSONString(parameterValues[i]));                }            }            messageCache.setOperationType(Integer.valueOf(cacheRefresh.operationType()));            //给定一个有序的版本号(默认统一的工作ID和数据中心ID)            messageCache.setVersion(SnowflakeIdWorker.getVersion());            messageCache.setMessageType(Integer.valueOf(cacheRefresh.messageType()));            messageCache.setCacheType(Integer.valueOf(cacheRefresh.cacheType()));            messageCache.setCreateDate(new Date());                        //将缓存数据写入读写队列            //缓存数据写入读写队列后，会定时每秒批量写入数据库(缓存数据写入DB只用于兜底，所以偶尔出现丢失并不影响)            DataRefreshDetailDO dataRefreshDetailDO = cacheRefreshConverter.converter(messageCache);            cacheQueue.submit(dataRefreshDetailDO);                         //发送MQ消息去处理缓存数据，比如将缓存数据更新到缓存上            //一般来说，热点缓存会比普通缓存少很多，所以普通缓存的更新会比较多，热点缓存的更新会比较少            //此外，热点缓存的更新会对时效性要求比较高，通过消息去异步处理本来就已存在一定的延迟            //所以这里将普通缓存和热点缓存的更新进行分开处理，减少时效性高的消息的处理延迟            if (CacheConstant.MESSAGE_TYPE_HOT.equals(cacheRefresh.messageType())) {                producer.sendMessage(RocketMqConstant.DATA_HOT_RADIO_TOPIC, JSONObject.toJSONString(messageCache), "热点缓存消息发送");            } else {                producer.sendMessage(RocketMqConstant.DATA_MESSAGE_CACHE_TOPIC, JSONObject.toJSONString(messageCache), cacheRefresh.mqCacheKey(), "通用缓存消息发送");            }        } catch (Exception e) {            log.error("处理缓存同步:{}失败，异常信息:{}", methodSignature.getMethod().getName(), e);        }        return response;    }}
//消息缓存处理对象@Datapublic class MessageCache implements Serializable {    //缓存的key    private String cacheKey;    //缓存的操作类型    private Integer operationType;    //使用的缓存类型，1为Redis，2为Tair    private Integer cacheType;    //缓存的消息内容    private String cacheJSON;    //消息的版本号(默认用时间戳来标记先后顺序)    private String version;    //缓存的数据状态是否还有效(0默认有效，1无效)    private Integer cacheStatus = 0;    //消息的创建时间    private Date createDate;    //是否热点消息，0普通消息，1热点消息    private Integer messageType = 0;}
//外部消息处理对象@Data@TableName("data_refresh_detail")public class DataRefreshDetailDO extends BaseEntity {    //缓存的key    private String cacheKey;    //缓存的操作类型    private Integer operationType = 1;    //使用的缓存类型，1为Redis，2为Tair    private Integer cacheType;    //缓存的消息内容    private String cacheJSON;    //消息的版本号(默认用时间戳来标记先后顺序)    private String version;    //是否热点消息，0普通消息，1热点消息    private Integer messageType = 0;}
//缓存数据的读写队列@Componentpublic class CacheQueue {    //提供锁的实例对象    private final PutDataLock lock = new PutDataLock();    //缓存数据的写队列    private volatile List<DataRefreshDetailDO> writeQueue = new LinkedList<>();    //缓存数据的读队列    private volatile List<DataRefreshDetailDO> readQueue = new LinkedList<>();    //是否正在写入数据    private volatile boolean isWrite = false;    ...        //缓存数据写入写队列    //@param dataRefreshDetailDO db存储对象    public void submit(DataRefreshDetailDO dataRefreshDetailDO) {        lock.lock();        try {            writeQueue.add(dataRefreshDetailDO);        } finally {            lock.unlock();        }    }    ...}
//锁竞争类对象//由于使用这个自旋锁是用于处理内存操作的，所以会很快处理完，可以忽略CPU的消耗public class PutDataLock {    private final AtomicBoolean putMessageSpinLock = new AtomicBoolean(true);        //上锁    public void lock() {        boolean flag;        do {            flag = this.putMessageSpinLock.compareAndSet(true, false);        }        while (!flag);    }        //解锁    public void unlock() {        this.putMessageSpinLock.compareAndSet(false, true);    }}

//数据源配置@Componentpublic class DataSourceConfig extends AbstractDataSourceConfig {    private SqlSessionTemplate sqlSessionTemplate;//存储数据源对象        @Autowired    private CacheQueue cacheQueue;        //初始化加载目前需要进行数据源的相关配置    @PostConstruct    public void initMigrateDateSource() throws Exception {        //加载数据源        DataSource dataSource = buildDataSource();        SqlSessionFactoryBean sqlSessionFactory = new SqlSessionFactoryBean();        sqlSessionFactory.setDataSource(dataSource);        sqlSessionTemplate = new SqlSessionTemplate(sqlSessionFactory.getObject());          //启动一个定时任务触发写入缓存数据到DB，每隔1秒触发一次，避免每次有缓存请求都执行DB操作        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,            new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build()        );        executorService.scheduleAtFixedRate(new Runnable() {            @Override            public void run() {                //判断当前是否正在将读写队列中的缓存数据写入DB                if (!cacheQueue.getIsWrite()) {                    //提交写队列中的缓存数据，然后写入DB                    cacheQueue.doCommit();                }            }        }, 1000, 1000, TimeUnit.MILLISECONDS);    }        //获取数据源    public SqlSession getSqlSession() {        try {            return SqlSessionUtils.getSqlSession(sqlSessionTemplate.getSqlSessionFactory(), sqlSessionTemplate.getExecutorType(), sqlSessionTemplate.getPersistenceExceptionTranslator());        } catch (Exception e) {            log.error("加载数据源对应连接池失败", e);        }        return null;    }        //关闭sqlSession    public void closeSqlSession(SqlSession session) {        SqlSessionUtils.closeSqlSession(session, sqlSessionTemplate.getSqlSessionFactory());    }}
//数据源配置public abstract class AbstractDataSourceConfig {    //构建数据源    public DruidDataSource buildDataSource() throws IOException {        Properties prop = new Properties();        InputStream inputStream = this.getClass().getClassLoader().getResourceAsStream("cache.properties");        prop.load(inputStream);        inputStream.close();        DruidDataSource druidDataSource = DataSourceBuilder.create()            .type(DruidDataSource.class)            .driverClassName(prop.getProperty("datasource.driver-class-name"))            .url(prop.getProperty("datasource.url"))            .username(prop.getProperty("datasource.username"))            .password(prop.getProperty("datasource.password"))            .build();        druidDataSource.setTestOnBorrow(true);        druidDataSource.setTestWhileIdle(true);        return druidDataSource;    }}
//缓存数据的读写队列@Componentpublic class CacheQueue {    //提供锁的实例对象    private final PutDataLock lock = new PutDataLock();    //缓存数据的写队列    private volatile List<DataRefreshDetailDO> writeQueue = new LinkedList<>();    //缓存数据的读队列    private volatile List<DataRefreshDetailDO> readQueue = new LinkedList<>();    //是否正在写入数据    private volatile boolean isWrite = false;    @Autowired    private DataRefreshService dataRefreshService;    ...        //交换读写队列    private void swapRequests() {        lock.lock();        try {            List<DataRefreshDetailDO> tmp = writeQueue;            writeQueue = readQueue;            readQueue = tmp;        } finally {            lock.unlock();        }    }        //提交写队列中的缓存数据，然后写入DB    public void doCommit() {        this.isWrite = true;        //交互读写队列后，再将读队列中的缓存数据写入DB        swapRequests();        if (!readQueue.isEmpty()) {            //先进行数据切割，每次写入DB的记录为500条            List<List<DataRefreshDetailDO>> dataRefreshDetailList = DataCuttingUtil.dataCuttingString(readQueue, CollectionSize.WRITE_SIZE);            for (List<DataRefreshDetailDO> dataRefreshDetailDOS : dataRefreshDetailList) {                dataRefreshService.saveDataRefreshDetailList(dataRefreshDetailDOS);            }        }        readQueue.clear();        this.isWrite = false;    }        //每隔1秒执行的定时任务会调用这个方法，判断当前是否正在将读写队列中的缓存数据写入DB    //@return 是否正在读取    public Boolean getIsWrite() {        return this.isWrite;    }    ...}
//将缓存数据写入DB@Servicepublic class DataRefreshServiceImpl implements DataRefreshService {    @Resource    private DataSourceConfig dataSourceConfig;        @Override    public void saveDataRefreshDetailList(List<DataRefreshDetailDO> dataRefreshDetailDOList) {        SqlSession session = null;        PreparedStatement pst = null;        try {            StringBuffer sql = new StringBuffer();            sql.append("INSERT INTO data_refresh_detail(cache_key, operation_type, cache_json, version, message_type, cache_type, create_time, update_time) values (?,?,?,?,?,?,now(),now())");            session = dataSourceConfig.getSqlSession();            pst = session.getConnection().prepareStatement(sql.toString());            for (DataRefreshDetailDO dataRefreshDetailDO : dataRefreshDetailDOList) {                pst.setString(1, dataRefreshDetailDO.getCacheKey());                pst.setInt(2, dataRefreshDetailDO.getOperationType());                pst.setString(3, dataRefreshDetailDO.getCacheJSON());                pst.setString(4, dataRefreshDetailDO.getVersion());                pst.setInt(5, dataRefreshDetailDO.getMessageType());                pst.setInt(6, dataRefreshDetailDO.getCacheType());                pst.addBatch();            }            pst.executeBatch();        } catch (Exception e) {            log.error("sql执行失败：{}", e);        } finally {            closeSqlSession(session, pst);        }    }        //关闭连接    private void closeSqlSession(SqlSession session, PreparedStatement pst) {        if (pst != null) {            try {                pst.close();            } catch (SQLException e) {                e.printStackTrace();            }        }        dataSourceConfig.closeSqlSession(session);    }}

//缓存消息处理@Componentpublic class DataMessageCacheListener implements MessageListenerConcurrently {    @Autowired    private CacheQueue cacheQueue;        @Autowired    private RedisCache redisCache;        //上次记录消费的时间    private long lastTimestamp = -1L;        @Override    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {        for (MessageExt messageExt : list) {            try {                String messageData = new String(messageExt.getBody());                log.info("DataMessageCacheListener缓存数据变更刷新，消息内容：{}", messageData);                MessageCache messageCache = JsonUtil.json2Object(messageData, MessageCache.class);                  //根据消息的缓存key，获取到对应的内存队列，分散队列提高处理效率，并保证单key的执行不会并发                BlockingQueue blockingQueue = cacheQueue.getBlockingQueue(messageCache.getCacheKey());                //将缓存消息添加到对应的内存队列中                blockingQueue.offer(messageCache);                  //记录最新的消费数据时间                setCacheRefreshTime();            } catch (Exception e) {                log.error("consume error, 缓存消息写入队列失败", e);                //本次消费失败，下次重新消费                return ConsumeConcurrentlyStatus.RECONSUME_LATER;            }        }        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;    }        //记录最新的消费时间，为避免无效的set操作，这里控制每秒最多执行一次set    private synchronized void setCacheRefreshTime() {        //获取到当前的时间，精确到秒        long timestamp = System.currentTimeMillis() / 1000;        //同一个时间则默认不处理        if (lastTimestamp == timestamp) {            return;        }        //标记最新MQ消息的接收时间        //如果写入DB的最新数据的创建时间，⽐消息的最新消费时间早1分钟以上        //则认为⽬前的消息消费出现问题，此时会有另外⼀个线程去查询出DB的数据来刷新缓存(兜底处理)        redisCache.set(CacheConstant.CACHE_ROCKET_TIME_KEY, DateFormatUtils.format(new Date(), DateConstant.DATE_TIME_FORMAT_PATTERN), 0);        lastTimestamp = timestamp;    }}

//处理缓存消息时使用的内存队列@Componentpublic class CacheQueue {    //处理缓存消息的内存队列    private final List<BlockingQueue> messageQueue = new ArrayList<>();
    //配置的消息内存队列数量    @Value("${message.queue-num}")    private Integer messageQueueNum;
    @PostConstruct    public void init() {        ExecutorService executors = Executors.newFixedThreadPool(messageQueueNum);        for (int i = 0; i < messageQueueNum; i++) {            //设置一个队列最大容纳数量            BlockingQueue blockingQueue = new ArrayBlockingQueue(150000);            messageQueue.add(blockingQueue);            //每个内存队列都对应线程池executors里的一个任务线程任务，该任务线程会处理添加到该内存队列中的缓存消息            executors.execute(new CacheConsistencyRunner(blockingQueue));        }    }
    //对消息的key进行hash处理，从而定位到具体的内存队列上    public BlockingQueue getBlockingQueue(String key) {        //先获取到传入的消息key对应的hash值        long hash = HashUtil.murMurHash(key.getBytes());        //计算出对应的内存队列        int c = (int) (hash %= messageQueue.size());        BlockingQueue blockingQueue = messageQueue.get(c);        return blockingQueue;    }}
//处理内存队列中缓存消息的线程public class CacheConsistencyRunner implements Runnable {    //缓存消息的失效时间，超过这个失效时间的历史消息不处理新增修改    private final static Integer failureTime = 60;        //存放缓存消息的内存队列    private BlockingQueue blockingQueue;    private RedisCache redisCache;    private TairCache tairCache;    private DefaultProducer defaultProducer;        public CacheConsistencyRunner(BlockingQueue blockingQueue) {        this.blockingQueue = blockingQueue;        this.redisCache = ApplicationContextUtil.getBean(RedisCache.class);        this.tairCache = ApplicationContextUtil.getBean(TairCache.class);        this.defaultProducer = ApplicationContextUtil.getBean(DefaultProducer.class);    }        //处理一个内存队列中的缓存消息的线程    @Override    public void run() {        try {            //TODO 对blockingQueue使用wait+notify实现内存队列为空时线程挂起            while (true) {                MessageCache cache = (MessageCache) blockingQueue.take();                //先判断是不是删除类型的缓存操作                if (MessageOperationEnum.DELETE.getCode().equals(cache.getOperationType())) {                    deleteCache(cache);                } else {                    //其它类型操作都是修改或者新增                    refreshCache(cache);                }            }        } catch (Exception e) {            e.printStackTrace();            log.error("处理缓存消息异常", e);        }    }    ...}

//处理内存队列中缓存消息的线程public class CacheConsistencyRunner implements Runnable {    //缓存消息的失效时间，超过这个失效时间的历史消息不处理新增修改    private final static Integer failureTime = 60;    //存放缓存消息的内存队列    private BlockingQueue blockingQueue;    private RedisCache redisCache;    private TairCache tairCache;    private DefaultProducer defaultProducer;        public CacheConsistencyRunner(BlockingQueue blockingQueue) {        this.blockingQueue = blockingQueue;        this.redisCache = ApplicationContextUtil.getBean(RedisCache.class);        this.tairCache = ApplicationContextUtil.getBean(TairCache.class);        this.defaultProducer = ApplicationContextUtil.getBean(DefaultProducer.class);    }    ...        //刷新缓存，或者新增缓存    private void refreshCache(MessageCache cache) {        log.info("开始处理新增或者修改缓存{}", JSONObject.toJSONString(cache));        //处理缓存之前，先看看这个消息的时间是多少，避免复活一些僵尸数据        Boolean isCache = DateFormatUtil.compareTo(cache.getCreateDate(), failureTime);        //如果这条消息的创建时间已经超过的有效期，那么视为无效消息不处理        if (!isCache) {            return;        }        //获取缓存        String cacheStr = getCache(cache.getCacheType(), cache.getCacheKey());        //当前没有缓存记录，直接覆盖一条        if (StringUtils.isEmpty(cacheStr)) {            setCache(cache.getCacheType(), cache.getCacheKey(), JSONObject.toJSONString(cache), 0);            send(cache);            return;        }        MessageCache messageCache = JsonUtil.json2Object(cacheStr, MessageCache.class);        //判断版本和缓存版本的记录谁更新，记录以最新的为准，避免低版本覆盖高版本        if (cache.getVersion().compareTo(messageCache.getVersion()) > 0) {            setCache(cache.getCacheType(), cache.getCacheKey(), JSONObject.toJSONString(cache), 0);            send(cache);        }    }        //获取缓存key    private String getCache(Integer cacheType, String cacheKey) {        if (CaCheTypeEnum.REDIS.getCode().equals(cacheType)) {            return redisCache.get(cacheKey);        }        return tairCache.get(cacheKey);    }        //设置缓存    private void setCache(Integer cacheType, String cacheKey, String cacheValue, Integer seconds) {        if (CaCheTypeEnum.REDIS.getCode().equals(cacheType)) {            redisCache.set(cacheKey, cacheValue, seconds);        } else {            tairCache.set(cacheKey, cacheValue, seconds);        }    }        //发送消息    public void send(MessageCache cache) {        //热点数据才处理发送广播消息        if (cache.getMessageType().equals(1)) {            defaultProducer.sendMessage(RocketMqConstant.DATA_HOT_RADIO_TOPIC, JSONObject.toJSONString(cache), 0, "广播消息发送");        }    }    ...}

//处理内存队列中缓存消息的线程public class CacheConsistencyRunner implements Runnable {    //缓存消息的失效时间，超过这个失效时间的历史消息不处理新增修改    private final static Integer failureTime = 60;    //存放缓存消息的内存队列    private BlockingQueue blockingQueue;    private RedisCache redisCache;    private TairCache tairCache;    private DefaultProducer defaultProducer;        public CacheConsistencyRunner(BlockingQueue blockingQueue) {        this.blockingQueue = blockingQueue;        this.redisCache = ApplicationContextUtil.getBean(RedisCache.class);        this.tairCache = ApplicationContextUtil.getBean(TairCache.class);        this.defaultProducer = ApplicationContextUtil.getBean(DefaultProducer.class);    }    ...        //删除缓存    private void deleteCache(MessageCache cache) {        //获取缓存        String cacheStr = getCache(cache.getCacheType(), cache.getCacheKey());        MessageCache messageCache = null;        if (StringUtils.isEmpty(cacheStr)) {            messageCache = new MessageCache();            messageCache.setVersion("-1");        } else {            messageCache = JsonUtil.json2Object(cacheStr, MessageCache.class);        }          //判断一下新的请求版本是否超过或者等于缓存的版本        //如果是则标记为已删除，同时缓存60分钟，避免缓存穿透以及可能的新增无效请求无版本可比对        if (cache.getVersion().compareTo(messageCache.getVersion()) >= 0) {            cache.setCacheStatus(1);            setCache(cache.getCacheType(), cache.getCacheKey(), JSONObject.toJSONString(messageCache), failureTime * 60);            send(cache);        }    }        //设置缓存    private void setCache(Integer cacheType, String cacheKey, String cacheValue, Integer seconds) {        if (CaCheTypeEnum.REDIS.getCode().equals(cacheType)) {            redisCache.set(cacheKey, cacheValue, seconds);        } else {            tairCache.set(cacheKey, cacheValue, seconds);        }    }        //发送消息    public void send(MessageCache cache) {        //热点数据才处理发送广播消息        if (cache.getMessageType().equals(1)) {            defaultProducer.sendMessage(RocketMqConstant.DATA_HOT_RADIO_TOPIC, JSONObject.toJSONString(cache), 0, "广播消息发送");        }    }    ...}
//消息缓存处理对象@Datapublic class MessageCache implements Serializable {    //缓存的key    private String cacheKey;    //缓存的操作类型    private Integer operationType;    //使用的缓存类型，1为Redis，2为Tair    private Integer cacheType;    //缓存的消息内容    private String cacheJSON;    //消息的版本号(默认用时间戳来标记先后顺序)    private String version;    //缓存的数据状态是否还有效(0默认有效，1无效)    private Integer cacheStatus = 0;    //消息的创建时间    private Date createDate;    //是否热点消息，0普通消息，1热点消息    private Integer messageType = 0;}

//负责定时检查消费是否发生异常而需要从DB查询缓存数据并刷新缓存@Componentpublic class DataRefreshTask {    @Autowired    private RedisLock redisLock;        @Autowired    private MessageService messageService;        //每分钟验证下是否触发缓存DB兜底    @Scheduled(fixedDelay = 60000)    void DataRefreshTask() {        boolean lock = redisLock.lock(CacheConstant.CACHE_LOCK_KEY);        try {            if (lock) {                messageService.outDataCacheRefresh();            }        } finally {            redisLock.unlock(CacheConstant.CACHE_LOCK_KEY);        }    }}

@Servicepublic class MessageServiceImpl implements MessageService {    @Autowired    private RedisCache redisCache;        @Autowired    private DataRefreshRepository dataRefreshRepository;        @Autowired    private CacheQueue cacheQueue;    ...        //从DB查询缓存数据并刷新缓存    @Override    public void outDataCacheRefresh() {        //1.先获取缓存中最新的消息消费时间，这里先减去1分钟，避免和MQ的处理直接重复        String createDate = redisCache.get(CacheConstant.CACHE_ROCKET_TIME_KEY);        if (StringUtils.isEmpty(createDate)) {            //如果缓存都不存在最新消费时间，默认处理1个小时内的DB数据，再超过前的缓存没有处理必要            createDate = DateFormatUtil.getHoursDate(-1);        } else {            //缓存时间存在，对时间减去一分钟进行处理            createDate = DateFormatUtil.getMinuteDate(createDate, -1);        }        //每次处理DB兜底的查询之前，先删除掉查询时间范围外的数据，避免数据一直写入导致数据量过大影响性能        dataRefreshRepository.deleteDataRefresh(DateFormatUtil.getHoursDate(-1));          //2.获取是否有超过的最新消费时间的数据落入DB(默认查询超过1分钟还未消费的数据)        int pageNum = 1;        //设置每次查询的数据量，最大为500        int pageSize = CollectionSize.WRITE_SIZE;        Page<DataRefreshDetailDO> page = new Page<>(pageNum, pageSize);        Page<DataRefreshDetailDO> pageResult = dataRefreshRepository.queryDataRefreshDetailDOList(page, createDate);        List<DataRefreshDetailDO> dataRefreshDetailList = pageResult.getRecords();        //将缓存数据写入内存队列进行处理        dataRefreshQueue(dataRefreshDetailList);        try {            while (pageNum <= pageResult.getTotal()) {                pageNum += 1;                page.setCurrent(pageNum);                pageResult = dataRefreshRepository.queryDataRefreshDetailDOList(page, createDate);                //将缓存数据写入内存队列进行处理                dataRefreshQueue(pageResult.getRecords());                //每次循环获取数据后，休眠20ms，避免对数据库造成太大压力                Thread.sleep(20);            }        } catch (InterruptedException e) {            throw new BaseBizException(ProductExceptionCode.PRODUCT_SQL);        }    }        //将数据写入队列进行处理    private void dataRefreshQueue(List<DataRefreshDetailDO> dataRefreshDetailList) {        if (!CollectionUtils.isEmpty(dataRefreshDetailList)) {            for (DataRefreshDetailDO dataRefreshDetailDO : dataRefreshDetailList) {                MessageCache messageCache = dataMessageConverter.converter(dataRefreshDetailDO);                BlockingQueue blockingQueue = cacheQueue.getBlockingQueue(messageCache.getCacheKey());                blockingQueue.offer(messageCache);            }        }    }    ...}
@Repositorypublic class DataRefreshRepository {    @Resource    private DataRefreshMapper dataRefreshMapper;       //获取得到消息最近消费时间的记录(默认查询大于这个时间点大于1分钟的数据)    public Page<DataRefreshDetailDO> queryDataRefreshDetailDOList(Page<DataRefreshDetailDO> page, String createDate) {        LambdaQueryWrapper<DataRefreshDetailDO> queryWrapper = Wrappers.lambdaQuery();        //查询创建时间大于这个版本号且未处理的数据        queryWrapper.gt(DataRefreshDetailDO::getVersion, createDate);        queryWrapper.eq(DataRefreshDetailDO::getCacheStatus, 0);        return dataRefreshMapper.selectPage(page, queryWrapper);    }        //删除超过一定时间区间的缓存DB数据    public void deleteDataRefresh(String createDate) {        LambdaUpdateWrapper<DataRefreshDetailDO> updateWrapper = Wrappers.lambdaUpdate();        updateWrapper.lt(DataRefreshDetailDO::getVersion, createDate);        dataRefreshMapper.delete(updateWrapper);    }}

@Retention(RetentionPolicy.RUNTIME)@Target(ElementType.METHOD)public @interface CacheRefresh {    //缓存消息key    String mqCacheKey();    //缓存的key    String cacheKey();    //需要缓存的值在方法参数中的坐标偏移量，默认不传取第一个参数    String index() default "0";    //是否热点缓存消息：0普通缓存消息，1热点缓存消息    String messageType() default "0";    //缓存的操作类型：1新增/修改，2删除    String operationType() default "1";    //缓存的组件类别：1是Redis，2是Tair    String cacheType() default "1";}
//刷新缓存的自定义注解@Aspect@Componentpublic class CacheRefreshAspect {    @Autowired    private DataRefreshProducer producer;        @Autowired    private CacheRefreshConverter cacheRefreshConverter;        @Autowired    private CacheQueue cacheQueue;
    //切入点，@CacheRefresh注解标注的    @Pointcut("@annotation(com.demo.eshop.cache.annotation.CacheRefresh)")    public void pointcut() {    }
    //环绕通知，在方法执行前后    //@param point 切入点    //@return 结果    @Around("pointcut() && @annotation(cacheRefresh)")    public Object around(ProceedingJoinPoint point, CacheRefresh cacheRefresh) throws Throwable {        //签名信息        Signature signature = point.getSignature();        //强转为方法信息        MethodSignature methodSignature = (MethodSignature) signature;        //参数名称        String[] parameterNames = methodSignature.getParameterNames();        //参数值        Object[] parameterValues = point.getArgs();        Object response;        try {            //先执行本地方法再执行异步的操作            response = point.proceed();        } catch (Throwable throwable) {            log.error("执行方法: {}失败，异常信息: {}", methodSignature.getMethod().getName(), throwable);            throw throwable;        }
        try {            MessageCache messageCache = new MessageCache();            for (int i = 0; i < parameterValues.length; i++) {                if (parameterNames[i].equals(cacheRefresh.cacheKey())) {                    messageCache.setCacheKey(String.valueOf(parameterValues[i]));                }                if (Integer.valueOf(cacheRefresh.index()) == i) {                    messageCache.setCacheJSON(JSONObject.toJSONString(parameterValues[i]));                }            }            messageCache.setOperationType(Integer.valueOf(cacheRefresh.operationType()));            //给定一个有序的版本号(默认统一的工作ID和数据中心ID)            messageCache.setVersion(SnowflakeIdWorker.getVersion());            messageCache.setMessageType(Integer.valueOf(cacheRefresh.messageType()));            messageCache.setCacheType(Integer.valueOf(cacheRefresh.cacheType()));            messageCache.setCreateDate(new Date());                        //将缓存数据写入读写队列            //缓存数据写入读写队列后，会定时每秒批量写入数据库(缓存数据写入DB只用于兜底，所以偶尔出现丢失并不影响)            DataRefreshDetailDO dataRefreshDetailDO = cacheRefreshConverter.converter(messageCache);            cacheQueue.submit(dataRefreshDetailDO);                         //发送MQ消息去处理缓存数据，比如将缓存数据更新到缓存上            //一般来说，热点缓存会比普通缓存少很多，所以普通缓存的更新会比较多，热点缓存的更新会比较少            //此外，热点缓存的更新会对时效性要求比较高，通过消息去异步处理本来就已存在一定的延迟            //所以这里将普通缓存和热点缓存的更新进行分开处理，减少时效性高的消息的处理延迟            if (CacheConstant.MESSAGE_TYPE_HOT.equals(cacheRefresh.messageType())) {                producer.sendMessage(RocketMqConstant.DATA_HOT_RADIO_TOPIC, JSONObject.toJSONString(messageCache), "热点缓存消息发送");            } else {                producer.sendMessage(RocketMqConstant.DATA_MESSAGE_CACHE_TOPIC, JSONObject.toJSONString(messageCache), cacheRefresh.mqCacheKey(), "通用缓存消息发送");            }        } catch (Exception e) {            log.error("处理缓存同步:{}失败，异常信息:{}", methodSignature.getMethod().getName(), e);        }        return response;    }}

//缓存数据的读写队列@Componentpublic class CacheQueue {    //提供锁的实例对象    private final PutDataLock lock = new PutDataLock();    //缓存数据的写队列    private volatile List<DataRefreshDetailDO> writeQueue = new LinkedList<>();    //缓存数据的读队列    private volatile List<DataRefreshDetailDO> readQueue = new LinkedList<>();    //是否正在写入数据    private volatile boolean isWrite = false;        @Autowired    private DataRefreshService dataRefreshService;        //缓存数据写入写队列    //@param dataRefreshDetailDO db存储对象    public void submit(DataRefreshDetailDO dataRefreshDetailDO) {        lock.lock();        try {            writeQueue.add(dataRefreshDetailDO);        } finally {            lock.unlock();        }    }        //交换读写队列    private void swapRequests() {        lock.lock();        try {            List<DataRefreshDetailDO> tmp = writeQueue;            writeQueue = readQueue;            readQueue = tmp;        } finally {            lock.unlock();        }    }        //提交写队列中的缓存数据，然后写入DB    public void doCommit() {        this.isWrite = true;        //交互读写队列后，再将读队列中的缓存数据写入DB        swapRequests();        if (!readQueue.isEmpty()) {            //先进行数据切割，每次写入DB的记录为500条            List<List<DataRefreshDetailDO>> dataRefreshDetailList = DataCuttingUtil.dataCuttingString(readQueue, CollectionSize.WRITE_SIZE);            for (List<DataRefreshDetailDO> dataRefreshDetailDOS : dataRefreshDetailList) {                dataRefreshService.saveDataRefreshDetailList(dataRefreshDetailDOS);            }        }        readQueue.clear();        this.isWrite = false;    }        //每隔1秒执行的定时任务会调用这个方法，判断当前是否正在将读写队列中的缓存数据写入DB    //@return 是否正在读取    public Boolean getIsWrite() {        return this.isWrite;    }}
//数据源配置@Componentpublic class DataSourceConfig extends AbstractDataSourceConfig {    private SqlSessionTemplate sqlSessionTemplate;//存储数据源对象        @Autowired    private CacheQueue cacheQueue;        //初始化加载目前需要进行数据源的相关配置    @PostConstruct    public void initMigrateDateSource() throws Exception {        //加载数据源        DataSource dataSource = buildDataSource();        SqlSessionFactoryBean sqlSessionFactory = new SqlSessionFactoryBean();        sqlSessionFactory.setDataSource(dataSource);        sqlSessionTemplate = new SqlSessionTemplate(sqlSessionFactory.getObject());          //启动一个定时任务触发写入缓存数据到DB，每隔1秒触发一次，避免每次有缓存请求都执行DB操作        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,            new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build()        );        executorService.scheduleAtFixedRate(new Runnable() {            @Override            public void run() {                //判断当前是否正在将读写队列中的缓存数据写入DB                if (!cacheQueue.getIsWrite()) {                    //提交写队列中的缓存数据，然后写入DB                    cacheQueue.doCommit();                }            }        }, 1000, 1000, TimeUnit.MILLISECONDS);    }    ...}

//缓存消息处理@Componentpublic class DataMessageCacheListener implements MessageListenerConcurrently {    @Autowired    private CacheQueue cacheQueue;        @Autowired    private RedisCache redisCache;        //上次记录消费的时间    private long lastTimestamp = -1L;        @Override    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {        for (MessageExt messageExt : list) {            try {                String messageData = new String(messageExt.getBody());                log.info("DataMessageCacheListener缓存数据变更刷新，消息内容：{}", messageData);                MessageCache messageCache = JsonUtil.json2Object(messageData, MessageCache.class);                  //根据消息的缓存key，获取到对应的内存队列，分散队列提高处理效率，并保证单key的执行不会并发                BlockingQueue blockingQueue = cacheQueue.getBlockingQueue(messageCache.getCacheKey());                //将缓存消息添加到对应的内存队列中                blockingQueue.offer(messageCache);                  //记录最新的消费数据时间                setCacheRefreshTime();            } catch (Exception e) {                log.error("consume error, 缓存消息写入队列失败", e);                //本次消费失败，下次重新消费                return ConsumeConcurrentlyStatus.RECONSUME_LATER;            }        }        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;    }    ...}
//处理缓存消息时使用的内存队列@Componentpublic class CacheQueue {    //处理缓存消息的内存队列    private final List<BlockingQueue> messageQueue = new ArrayList<>();
    //配置的消息内存队列数量    @Value("${message.queue-num}")    private Integer messageQueueNum;
    @PostConstruct    public void init() {        ExecutorService executors = Executors.newFixedThreadPool(messageQueueNum);        for (int i = 0; i < messageQueueNum; i++) {            //设置一个队列最大容纳数量            BlockingQueue blockingQueue = new ArrayBlockingQueue(150000);            messageQueue.add(blockingQueue);            //每个内存队列都对应线程池executors里的一个任务线程任务，该任务线程会处理添加到该内存队列中的缓存消息            executors.execute(new CacheConsistencyRunner(blockingQueue));        }    }
    //对消息的key进行hash处理，从而定位到具体的内存队列上    public BlockingQueue getBlockingQueue(String key) {        //先获取到传入的消息key对应的hash值        long hash = HashUtil.murMurHash(key.getBytes());        //计算出对应的内存队列        int c = (int) (hash %= messageQueue.size());        BlockingQueue blockingQueue = messageQueue.get(c);        return blockingQueue;    }}
//处理内存队列中缓存消息的线程public class CacheConsistencyRunner implements Runnable {    //缓存消息的失效时间，超过这个失效时间的历史消息不处理新增修改    private final static Integer failureTime = 60;        //存放缓存消息的内存队列    private BlockingQueue blockingQueue;        private RedisCache redisCache;    private TairCache tairCache;    private DefaultProducer defaultProducer;        public CacheConsistencyRunner(BlockingQueue blockingQueue) {        this.blockingQueue = blockingQueue;        this.redisCache = ApplicationContextUtil.getBean(RedisCache.class);        this.tairCache = ApplicationContextUtil.getBean(TairCache.class);        this.defaultProducer = ApplicationContextUtil.getBean(DefaultProducer.class);    }        //处理一个内存队列中的缓存消息的线程    @Override    public void run() {        try {            //TODO 对blockingQueue使用wait+notify实现内存队列为空时线程挂起            while (true) {                MessageCache cache = (MessageCache) blockingQueue.take();                //先判断是不是删除类型的缓存操作                if (MessageOperationEnum.DELETE.getCode().equals(cache.getOperationType())) {                    deleteCache(cache);                } else {                    //其它类型操作都是修改或者新增                    refreshCache(cache);                }            }        } catch (Exception e) {            e.printStackTrace();            log.error("处理缓存消息异常", e);        }    }        //删除缓存    private void deleteCache(MessageCache cache) {        //获取缓存        String cacheStr = getCache(cache.getCacheType(), cache.getCacheKey());        MessageCache messageCache = null;        if (StringUtils.isEmpty(cacheStr)) {            messageCache = new MessageCache();            messageCache.setVersion("-1");        } else {            messageCache = JsonUtil.json2Object(cacheStr, MessageCache.class);        }          //判断一下新的请求版本是否超过或者等于缓存的版本        //如果是则标记为已删除，同时缓存60分钟，避免缓存穿透以及可能的新增无效请求无版本可比对        if (cache.getVersion().compareTo(messageCache.getVersion()) >= 0) {            cache.setCacheStatus(1);            setCache(cache.getCacheType(), cache.getCacheKey(), JSONObject.toJSONString(messageCache), failureTime * 60);            send(cache);        }    }        //刷新缓存，或者新增缓存    private void refreshCache(MessageCache cache) {        log.info("开始处理新增或者修改缓存{}", JSONObject.toJSONString(cache));        //处理缓存之前，先看看这个消息的时间是多少，避免复活一些僵尸数据        Boolean isCache = DateFormatUtil.compareTo(cache.getCreateDate(), failureTime);        //如果这条消息的创建时间已经超过的有效期，那么视为无效消息不处理        if (!isCache) {            return;        }        //获取缓存        String cacheStr = getCache(cache.getCacheType(), cache.getCacheKey());        //当前没有缓存记录，直接覆盖一条        if (StringUtils.isEmpty(cacheStr)) {            setCache(cache.getCacheType(), cache.getCacheKey(), JSONObject.toJSONString(cache), 0);            send(cache);            return;        }        MessageCache messageCache = JsonUtil.json2Object(cacheStr, MessageCache.class);        //判断版本和缓存版本的记录谁更新，记录以最新的为准，避免低版本覆盖高版本        if (cache.getVersion().compareTo(messageCache.getVersion()) > 0) {            setCache(cache.getCacheType(), cache.getCacheKey(), JSONObject.toJSONString(cache), 0);            send(cache);        }    }    ...}