Redis面试题记录--缓存双写情况下导致数据不一致问题

转载自：https://blog.csdn.net/lzhcoder/article/details/79469123

https://blog.csdn.net/u013374645/article/details/91409150

1.最经典的缓存+数据库读写的模式，cache aside pattern

1.1、Cache Aside Pattern

（1）读的时候，先读缓存，缓存没有的话，那么就读数据库，然后取出数据后放入缓存，同时返回响应

（2）更新的时候，先删除缓存，然后再更新数据库

1.2、为什么是删除缓存，而不是更新缓存呢？

（1）、如果写数据库的值与更新到缓存值是一样的，不需要经过任何的计算，可以马上更新缓存，但是如果对于那种写数据频繁而读数据少的场景并不合适这种解决方案，因为也许还没有查询就被删除或修改了，这样会浪费时间和资源

（2）、如果写数据库的值与更新缓存的值不一致，写入缓存中的数据需要经过几个表的关联计算后得到的结果插入缓存中，那就没有必要马上更新缓存，只要删除缓存即可，等到查询的时候在去把计算后得到的结果插入到缓存中即可。其实删除缓存，而不是更新缓存，就是一个lazy计算的思想。

1.3、先删除缓存，再更新数据库 && 先更新数据库，再删除缓存的比较

第一种方案的分析见2.1和2.2

下面讨论第二种方案，该设计模式产生双写不一致的可能情况：

①一个是读操作，但是没有命中缓存，然后就到数据库中取数据，此时来了一个写操作，写完数据库后，让缓存失效，然后，之前的那个读操作再把老的数据放进去，所以，会造成脏数据。

该情况出现的概率可能非常低，因为这个条件需要发生在读缓存时缓存失效，而且并发着有一个写操作。而实际上数据库的写操作会比读操作慢得多，而且还要锁表，而读操作必需在写操作前进入数据库操作，而又要晚于写操作更新缓存，所有的这些条件都具备的概率基本并不大。

②删除缓存失败

2.在库存服务中实现缓存与数据库双写一致性保障方案实战

实时性比较高的数据缓存，选择的就是库存的服务

库存可能会修改，每次修改都要去更新这个缓存数据; 每次库存的数据，在缓存中一旦过期，或者是被清理掉了，前端的nginx服务都会发送请求给库存服务，去获取相应的数据

库存这一块，写数据库的时候，直接更新redis缓存

实际上没有这么的简单，这里，其实就涉及到了一个问题，数据库与缓存双写，数据不一致的问题

围绕和结合实时性较高的库存服务，把数据库与缓存双写不一致问题以及其解决方案，给大家讲解一下

数据库与缓存双写不一致，很常见的问题，大型的缓存架构中，第一个解决方案

2.1、最初级的缓存不一致问题以及解决方案

问题：先修改数据库，再删除缓存，如果删除缓存失败了，那么会导致数据库中是新数据，缓存中是旧数据，数据出现不一致

解决思路：

先删除缓存，再修改数据库。如果删除缓存成功，修改数据库失败了，那么数据库中是旧数据，缓存中是空的，那么数据不会不一致，如果删除缓存失败不执行修改数据库（cache aside pattern）

2.2、比较复杂的数据不一致问题分析

数据发生了变更，先删除了缓存，然后要去修改数据库，此时还没修改

一个请求过来，去读缓存，发现缓存空了，去查询数据库，查到了修改前的旧数据，放到了缓存中

数据变更的程序完成了数据库的修改

完了，数据库和缓存中的数据不一样了。。。。

2.3、为什么上亿流量高并发场景下，缓存会出现这个问题？

只有在对一个数据在并发的进行读写的时候，才可能会出现这种问题

其实如果说你的并发量很低的话，特别是读并发很低，每天访问量就1万次，那么很少的情况下，会出现刚才描述的那种不一致的场景

但是问题是，如果每天的是上亿的流量，每秒并发读是几万，每秒只要有数据更新的请求，就可能会出现上述的数据库+缓存不一致的情况

高并发了以后，问题是很多的

2.4、数据库与缓存更新与读取操作进行异步串行化

更新数据的时候(写请求)，根据数据的唯一标识，将操作路由之后，发送到AarrayBlockQueue中

读取数据的时候，如果发现数据不在缓存中，那么将读取mysql数据+更新缓存的操作(读请求)，根据产品id路由之后，也发送同一个AarrayBlockQueue中

一个队列对应一个工作线程,每个工作线程串行拿到对应的操作，然后一条一条的执行

这样的话，一个数据变更的操作，先执行，删除缓存，然后再去更新数据库，但是还没完成更新

此时如果一个读请求过来，读到了空的缓存，那么可以先将缓存更新的请求发送到队列中，此时会在队列中积压，然后同步等待缓存更新完成

这里有一个优化点，一个队列中同一个商品，每次调用获取库存的接口时，都会发起一个读取mysql数据+更新缓存的异步请求，这些请求串在一起是没意义的，因此可以做过滤，如果发现队列中同一个商品已经有一个更新缓存的请求了，那么就不用再放这个商品的更新请求操作进去了，直接等待前面的更新操作请求完成即可（代码中利用flagMap实现过滤）

注解：同一个商品在同一个内存队列中，保证删除缓存，和修改数据库的操作不会有其他线程干扰

如果请求还在等待时间范围内，不断轮询发现可以取到值了，那么就直接返回; 如果请求等待的时间超过一定时长，那么这一次直接从数据库中读取当前的旧值

2.4.1、线程池+内存队列初始化

/**注册监听器**/

@Bean
public ServletListenerRegistrationBean servletListenerRegistrationBean(){
    ServletListenerRegistrationBean servletListenerRegistrationBean = new ServletListenerRegistrationBean();
    servletListenerRegistrationBean.setListener(new InitListener());
    return servletListenerRegistrationBean;
}

　　

import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;

import com.roncoo.eshop.inventory.thread.RequestProcessorThreadPool;

/**
 * 系统初始化监听器
 * @author Administrator
 *
 */
public class InitListener implements ServletContextListener {

    @Override
    public void contextInitialized(ServletContextEvent sce) {
        // 初始化工作线程池和内存队列
        RequestProcessorThreadPool.init();
    }

    @Override
    public void contextDestroyed(ServletContextEvent sce) {

    }

}

　　

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.roncoo.eshop.inventory.request.Request;
import com.roncoo.eshop.inventory.request.RequestQueue;

/**
 * 请求处理线程池：单例
 * @author Administrator
 *
 */
public class RequestProcessorThreadPool {

    // 在实际项目中，你设置线程池大小是多少，每个线程监控的那个内存队列的大小是多少
    // 都可以做到一个外部的配置文件中
    // 我们这了就给简化了，直接写死了，好吧

    /**
     * 线程池
     */
    private ExecutorService threadPool = Executors.newFixedThreadPool(10);

    public RequestProcessorThreadPool() {
        RequestQueue requestQueue = RequestQueue.getInstance();

        for(int i = 0; i < 10; i++) {
            ArrayBlockingQueue<Request> queue = new ArrayBlockingQueue<Request>(100);
            requestQueue.addQueue(queue);
            threadPool.submit(new RequestProcessorThread(queue));
        }
    }

    /**
     * 单例有很多种方式去实现：我采取绝对线程安全的一种方式
     *
     * 静态内部类的方式，去初始化单例
     *
     * @author Administrator
     *
     */
    private static class Singleton {

        private static RequestProcessorThreadPool instance;

        static {
            instance = new RequestProcessorThreadPool();
        }

        public static RequestProcessorThreadPool getInstance() {
            return instance;
        }

    }

    /**
     * jvm的机制去保证多线程并发安全
     *
     * 内部类的初始化，一定只会发生一次，不管多少个线程并发去初始化
     *
     * @return
     */
    public static RequestProcessorThreadPool getInstance() {
        return Singleton.getInstance();
    }

    /**
     * 初始化的便捷方法
     */
    public static void init() {
        getInstance();
    }

}

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 请求内存队列
 * @author Administrator
 *
 */
public class RequestQueue {

    /**
     * 内存队列
     */
    private List<ArrayBlockingQueue<Request>> queues =
            new ArrayList<ArrayBlockingQueue<Request>>();
    /**
     * 标识位map
     */
    private Map<Integer, Boolean> flagMap = new ConcurrentHashMap<Integer, Boolean>();

    /**
     * 单例有很多种方式去实现：我采取绝对线程安全的一种方式
     *
     * 静态内部类的方式，去初始化单例
     *
     * @author Administrator
     *
     */
    private static class Singleton {

        private static RequestQueue instance;

        static {
            instance = new RequestQueue();
        }

        public static RequestQueue getInstance() {
            return instance;
        }

    }

    /**
     * jvm的机制去保证多线程并发安全
     *
     * 内部类的初始化，一定只会发生一次，不管多少个线程并发去初始化
     *
     * @return
     */
    public static RequestQueue getInstance() {
        return Singleton.getInstance();
    }

    /**
     * 添加一个内存队列
     * @param queue
     */
    public void addQueue(ArrayBlockingQueue<Request> queue) {
        this.queues.add(queue);
    }

    /**
     * 获取内存队列的数量
     * @return
     */
    public int queueSize() {
        return queues.size();
    }

    /**
     * 获取内存队列
     * @param index
     * @return
     */
    public ArrayBlockingQueue<Request> getQueue(int index) {
        return queues.get(index);
    }

    public Map<Integer, Boolean> getFlagMap() {
        return flagMap;
    }

}

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;

import com.roncoo.eshop.inventory.request.Request;

/**
 * 执行请求的工作线程
 * @author Administrator
 *
 */
public class RequestProcessorThread implements Callable<Boolean> {

    /**
     * 自己监控的内存队列
     */
    private ArrayBlockingQueue<Request> queue;

    public RequestProcessorThread(ArrayBlockingQueue<Request> queue) {
        this.queue = queue;
    }

    @Override
    public Boolean call() throws Exception {
        try {
            while(true) {
                // ArrayBlockingQueue
                // Blocking就是说明，如果队列满了，或者是空的，那么都会在执行操作的时候，阻塞住
                Request request = queue.take();
                System.out.println("===========日志===========: 工作线程处理请求，商品id=" + request.getProductId());
                // 执行这个request操作
                request.process();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

}

2.4.2、两种请求对象封装

import com.roncoo.eshop.inventory.model.ProductInventory;
import com.roncoo.eshop.inventory.service.ProductInventoryService;

/**
 * 重新加载商品库存的缓存
 * @author Administrator
 *
 */
public class ProductInventoryCacheRefreshRequest implements Request {

    /**
     * 商品id
     */
    private Integer productId;
    /**
     * 商品库存Service
     */
    private ProductInventoryService productInventoryService;

    public ProductInventoryCacheRefreshRequest(Integer productId,
            ProductInventoryService productInventoryService) {
        this.productId = productId;
        this.productInventoryService = productInventoryService;
    }

    @Override
    public void process() {
        // 从数据库中查询最新的商品库存数量
        ProductInventory productInventory = productInventoryService.findProductInventory(productId);
        System.out.println("===========日志===========: 已查询到商品最新的库存数量，商品id=" + productId + ", 商品库存数量=" + productInventory.getInventoryCnt());
        // 将最新的商品库存数量，刷新到redis缓存中去
        productInventoryService.setProductInventoryCache(productInventory);
    }

    public Integer getProductId() {
        return productId;
    }

}

import com.roncoo.eshop.inventory.model.ProductInventory;
import com.roncoo.eshop.inventory.service.ProductInventoryService;

/**
 * 比如说一个商品发生了交易，那么就要修改这个商品对应的库存
 *
 * 此时就会发送请求过来，要求修改库存，那么这个可能就是所谓的data update request，数据更新请求
 *
 * cache aside pattern
 *
 * （1）删除缓存
 * （2）更新数据库
 *
 * @author Administrator
 *
 */
public class ProductInventoryDBUpdateRequest implements Request {

    /**
     * 商品库存
     */
    private ProductInventory productInventory;
    /**
     * 商品库存Service
     */
    private ProductInventoryService productInventoryService;

    public ProductInventoryDBUpdateRequest(ProductInventory productInventory,
            ProductInventoryService productInventoryService) {
        this.productInventory = productInventory;
        this.productInventoryService = productInventoryService;
    }

    @Override
    public void process() {
        System.out.println("===========日志===========: 数据库更新请求开始执行，商品id=" + productInventory.getProductId() + ", 商品库存数量=" + productInventory.getInventoryCnt());
        // 删除redis中的缓存
        productInventoryService.removeProductInventoryCache(productInventory);
        // 为了模拟演示先删除了redis中的缓存，然后还没更新数据库的时候，读请求过来了，这里可以人工sleep一下
//        try {
//            Thread.sleep(20000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        // 修改数据库中的库存
        productInventoryService.updateProductInventory(productInventory);
    }

    /**
     * 获取商品id
     */
    public Integer getProductId() {
        return productInventory.getProductId();
    }

}

2.4.3、请求异步执行Service封装

2.4.4、读请求去重优化

对一个商品的库存的数据库更新操作已经在内存队列中了

然后对这个商品的库存的读取操作，要求读取数据库的库存数据，然后更新到缓存中，多个读

这多个读，其实只要有一个读请求操作压到队列里就可以了

其他的读操作，全部都wait那个读请求的操作，刷新缓存，就可以读到缓存中的最新数据

2.4.5、空数据读请求过滤优化

如果读请求发现redis缓存中没有数据，就会发送读请求给库存服务，但是此时缓存中为空，可能是因为写请求先删除了缓存，也可能是数据库里压根儿没这条数据

如果是数据库中压根儿没这条数据的场景，那么就不应该将读请求操作给压入队列中，而是直接返回空就可以了

都是为了减少内存队列中的请求积压，内存队列中积压的请求越多，就可能导致每个读请求hang住的时间越长，也可能导致多个读请求被hang住

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;

import org.springframework.stereotype.Service;

import com.roncoo.eshop.inventory.request.ProductInventoryCacheRefreshRequest;
import com.roncoo.eshop.inventory.request.ProductInventoryDBUpdateRequest;
import com.roncoo.eshop.inventory.request.Request;
import com.roncoo.eshop.inventory.request.RequestQueue;
import com.roncoo.eshop.inventory.service.RequestAsyncProcessService;

/**
 * 请求异步处理的service实现
 * @author Administrator
 *
 */
@Service("requestAsyncProcessService")
public class RequestAsyncProcessServiceImpl implements RequestAsyncProcessService {

    @Override
    public void process(Request request) {
        try {
            // 先做读请求的去重
            RequestQueue requestQueue = RequestQueue.getInstance();
            Map<Integer, Boolean> flagMap = requestQueue.getFlagMap();

            if(request instanceof ProductInventoryDBUpdateRequest) {
                // 如果是一个更新数据库的请求，那么就将那个productId对应的标识设置为true
                flagMap.put(request.getProductId(), true);
            } else if(request instanceof ProductInventoryCacheRefreshRequest) {
                Boolean flag = flagMap.get(request.getProductId());

                // 如果flag是null
                if(flag == null) {
                    flagMap.put(request.getProductId(), false);
                }

                // 如果是缓存刷新的请求，那么就判断，如果标识不为空，而且是true，就说明之前有一
                                //个这个商品的数据库更新请求
                if(flag != null && flag) {
                    flagMap.put(request.getProductId(), false);
                }

                // 如果是缓存刷新的请求，而且发现标识不为空，但是标识是false
                // 说明前面已经有一个缓存刷新请求了
                if(flag != null && !flag) {
                    // 对于这种读请求，直接就过滤掉，不要放到后面的内存队列里面去了
                    return;
                }
            }

            // 做请求的路由，根据每个请求的商品id，路由到对应的内存队列中去
            ArrayBlockingQueue<Request> queue = getRoutingQueue(request.getProductId());
            // 将请求放入对应的队列中，完成路由操作
            queue.put(request);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取路由到的内存队列
     * @param productId 商品id
     * @return 内存队列
     */
    private ArrayBlockingQueue<Request> getRoutingQueue(Integer productId) {
        RequestQueue requestQueue = RequestQueue.getInstance();

        // 先获取productId的hash值
        String key = String.valueOf(productId);
        int h;
        int hash = (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);

        // 对hash值取模，将hash值路由到指定的内存队列中，比如内存队列大小8
        // 用内存队列的数量对hash值取模之后，结果一定是在0~7之间
        // 所以任何一个商品id都会被固定路由到同样的一个内存队列中去的
        int index = (requestQueue.queueSize() - 1) & hash;

        System.out.println("===========日志===========: 路由内存队列，商品id=" + productId + ", 队列索引=" + index);  

        return requestQueue.getQueue(index);
    }

}

2.4.6、两种请求Controller接口封装

如果请求还在等待时间范围内，不断轮询发现可以取到值了，那么就直接返回; 如果请求等待的时间超过一定时长，那么这一次直接从数据库中读取当前的旧值

    /**
     * 获取商品库存
     */
    @RequestMapping("/getProductInventory")
    @ResponseBody
    public ProductInventory getProductInventory(Integer productId) {
        System.out.println("===========日志===========: 接收到一个商品库存的读请求，商品id=" + productId);  

        ProductInventory productInventory = null;

        try {
            Request request = new ProductInventoryCacheRefreshRequest(
                    productId, productInventoryService);
            requestAsyncProcessService.process(request);

            // 将请求扔给service异步去处理以后，就需要while(true)一会儿，在这里hang住
            // 去尝试等待前面有商品库存更新的操作，同时缓存刷新的操作，将最新的数据刷新到缓存中
            long startTime = System.currentTimeMillis();
            long endTime = 0L;
            long waitTime = 0L;

            // 等待超过200ms没有从缓存中获取到结果
            while(true) {
//                if(waitTime > 25000) {
//                    break;
//                }

                // 一般公司里面，面向用户的读请求控制在200ms就可以了
                if(waitTime > 200) {
                    break;
                }

                // 尝试去redis中读取一次商品库存的缓存数据
                productInventory = productInventoryService.getProductInventoryCache(productId);

                // 如果读取到了结果，那么就返回
                if(productInventory != null) {
                    System.out.println("===========日志===========: 在200ms内读取到了redis中的库存缓存，商品id=" + productInventory.getProductId() + ", 商品库存数量=" + productInventory.getInventoryCnt());
                    return productInventory;
                }

                // 如果没有读取到结果，那么等待一段时间
                else {
                    Thread.sleep(20);
                    endTime = System.currentTimeMillis();
                    waitTime = endTime - startTime;
                }
            }

            // 直接尝试从数据库中读取数据
            productInventory = productInventoryService.findProductInventory(productId);
            if(productInventory != null) {
                // 将缓存刷新一下
                productInventoryService.setProductInventoryCache(productInventory);
                return productInventory;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        return new ProductInventory(productId, -1L);
    }

    /**
     * 更新商品库存
     */
    @RequestMapping("/updateProductInventory")
    @ResponseBody
    public Response updateProductInventory(ProductInventory productInventory) {
        // 为了简单起见，我们就不用log4j那种日志框架去打印日志了
        // 其实log4j也很简单，实际企业中都是用log4j去打印日志的，自己百度一下
        System.out.println("===========日志===========: 接收到更新商品库存的请求，商品id=" + productInventory.getProductId() + ", 商品库存数量=" + productInventory.getInventoryCnt());

        Response response = null;

        try {
            Request request = new ProductInventoryDBUpdateRequest(
                    productInventory, productInventoryService);
            requestAsyncProcessService.process(request);
            response = new Response(Response.SUCCESS);
        } catch (Exception e) {
            e.printStackTrace();
            response = new Response(Response.FAILURE);
        }

        return response;
    }

2.5、高并发的场景下，该解决方案要注意的问题

（1）读请求长时阻塞

由于读请求进行了非常轻度的异步化，所以一定要注意读超时的问题，每个读请求必须在超时时间范围内返回

该解决方案，最大的风险点在于说，可能数据更新很频繁，导致队列中积压了大量更新操作在里面，然后读请求会发生大量的超时，最后导致大量的请求直接走数据库

务必通过一些模拟真实的测试，看看更新数据的频繁是怎样的

另外一点，因为一个队列中，可能会积压针对多个数据项的更新操作，因此需要根据自己的业务情况进行测试，可能需要部署多个服务，每个服务分摊一些数据的更新操作

如果一个内存队列里居然会挤压100个商品的库存修改操作，每隔库存修改操作要耗费10ms区完成，那么最后一个商品的读请求，可能等待10 * 100 = 1000ms = 1s后，才能得到数据

这个时候就导致读请求的长时阻塞

一定要做根据实际业务系统的运行情况，去进行一些压力测试，和模拟线上环境，去看看最繁忙的时候，内存队列可能会挤压多少更新操作，可能会导致最后一个更新操作对应的读请求，会hang多少时间，如果读请求在200ms返回，如果你计算过后，哪怕是最繁忙的时候，积压10个更新操作，最多等待200ms，那还可以的

如果一个内存队列可能积压的更新操作特别多，那么你就要加机器，让每个机器上部署的服务实例处理更少的数据，那么每个内存队列中积压的更新操作就会越少

其实根据之前的项目经验，一般来说数据的写频率是很低的，因此实际上正常来说，在队列中积压的更新操作应该是很少的

针对读高并发，读缓存架构的项目，一般写请求相对读来说，是非常非常少的，每秒的QPS能到几百就不错了

一秒，500的写操作，5份，每200ms，就100个写操作

单机器，20个内存队列，每个内存队列，可能就积压5个写操作，每个写操作性能测试后，一般在20ms左右就完成

那么针对每个内存队列中的数据的读请求，也就最多hang一会儿，200ms以内肯定能返回了

写QPS扩大10倍，但是经过刚才的测算，就知道，单机支撑写QPS几百没问题，那么就扩容机器，扩容10倍的机器，10台机器，每个机器20个队列，200个队列

大部分的情况下，应该是这样的，大量的读请求过来，都是直接走缓存取到数据的

少量情况下，可能遇到读跟数据更新冲突的情况，如上所述，那么此时更新操作如果先入队列，之后可能会瞬间来了对这个数据大量的读请求，但是因为做了去重的优化，所以也就一个更新缓存的操作跟在它后面

等数据更新完了，读请求触发的缓存更新操作也完成，然后临时等待的读请求全部可以读到缓存中的数据

（2）读请求并发量过高

这里还必须做好压力测试，确保恰巧碰上上述情况的时候，还有一个风险，就是突然间大量读请求会在几十毫秒的延时hang在服务上，看服务能不能抗的住，需要多少机器才能抗住最大的极限情况的峰值

但是因为并不是所有的数据都在同一时间更新，缓存也不会同一时间失效，所以每次可能也就是少数数据的缓存失效了，然后那些数据对应的读请求过来，并发量应该也不会特别大

按1:99的比例计算读和写的请求，每秒5万的读QPS，可能只有500次更新操作

如果一秒有500的写QPS，那么要测算好，可能写操作影响的数据有500条，这500条数据在缓存中失效后，可能导致多少读请求，发送读请求到库存服务来，要求更新缓存

一般来说，1:1，1:2，1:3，每秒钟有1000个读请求，会hang在库存服务上，每个读请求最多hang多少时间，200ms就会返回

在同一时间最多hang住的可能也就是单机200个读请求，同时hang住

单机hang200个读请求，还是ok的

1:20，每秒更新500条数据，这500秒数据对应的读请求，会有20 * 500 = 1万

1万个读请求全部hang在库存服务上，就死定了

（3）多服务实例部署的请求路由

后端的nginx服务器，就称之为应用服务器; 最前端的nginx服务器，被称之为分发服务器

分发层nginx，负责流量分发的逻辑和策略，这个里面它可以根据你自己定义的一些规则，比如根据productId去进行hash，然后对后端的nginx数量取模

将某一个商品的访问的请求，就固定路由到一个后端的nginx服务器上去，固定商品id，走固定的缓存服务实例。保证说只会从redis中获取一次缓存数据，后面全都是走nginx本地缓存了

机器级别的请求路由问题

（4）热点商品的路由问题，导致请求的倾斜

万一某个商品的读写请求特别高，全部打到相同的机器的相同的队列里面去了，可能造成某台机器的压力过大

就是说，因为只有在商品数据更新的时候才会清空缓存，然后才会导致读写并发，所以更新频率不是太高的话，这个问题的影响并不是特别大

但是的确可能某些机器的负载会高一些

2.6、总结

一般来说，就是如果你的系统不是严格要求缓存+数据库必须一致性的话，缓存可以稍微的跟数据库偶尔有不一致的情况，最好不要做这个方案，读请求和写请求串行化，串到一个内存队列里去，这样就可以保证一定不会出现不一致的情况

串行化之后，就会导致系统的吞吐量会大幅度的降低，用比正常情况下多几倍的机器去支撑线上的一个请求。

最后附上代码地址https://gitee.com/lzhcode/eshop-inventory

2.4.1、线程池+内存队列初始化

1.  
   /**注册监听器**/
2.  
    
3.  
   @Bean
4.  
   public ServletListenerRegistrationBean servletListenerRegistrationBean(){
5.  
       ServletListenerRegistrationBean servletListenerRegistrationBean = new ServletListenerRegistrationBean();
6.  
       servletListenerRegistrationBean.setListener(new InitListener());
7.  
       return servletListenerRegistrationBean;
8.  
   }

1.  
   import javax.servlet.ServletContextEvent;
2.  
   import javax.servlet.ServletContextListener;
3.  
    
4.  
   import com.roncoo.eshop.inventory.thread.RequestProcessorThreadPool;
5.  
    
6.  
   /**
7.  
   * 系统初始化监听器
8.  
   * @author Administrator
9.  
   *
10.  
    */
11.  
    public class InitListener implements ServletContextListener {
12.  
     
13.  
    @Override
14.  
    public void contextInitialized(ServletContextEvent sce) {
15.  
    // 初始化工作线程池和内存队列
16.  
    RequestProcessorThreadPool.init();
17.  
    }
18.  
     
19.  
    @Override
20.  
    public void contextDestroyed(ServletContextEvent sce) {
21.  
     
22.  
    }
23.  
     
24.  
    }

1.  
   import java.util.concurrent.ArrayBlockingQueue;
2.  
   import java.util.concurrent.ExecutorService;
3.  
   import java.util.concurrent.Executors;
4.  
    
5.  
   import com.roncoo.eshop.inventory.request.Request;
6.  
   import com.roncoo.eshop.inventory.request.RequestQueue;
7.  
    
8.  
   /**
9.  
   * 请求处理线程池：单例
10.  
    * @author Administrator
11.  
    *
12.  
    */
13.  
    public class RequestProcessorThreadPool {
14.  
     
15.  
    // 在实际项目中，你设置线程池大小是多少，每个线程监控的那个内存队列的大小是多少
16.  
    // 都可以做到一个外部的配置文件中
17.  
    // 我们这了就给简化了，直接写死了，好吧
18.  
     
19.  
    /**
20.  
    * 线程池
21.  
    */
22.  
    private ExecutorService threadPool = Executors.newFixedThreadPool(10);
23.  
     
24.  
    public RequestProcessorThreadPool() {
25.  
    RequestQueue requestQueue = RequestQueue.getInstance();
26.  
     
27.  
    for(int i = 0; i < 10; i++) {
28.  
    ArrayBlockingQueue<Request> queue = new ArrayBlockingQueue<Request>(100);
29.  
    requestQueue.addQueue(queue);
30.  
    threadPool.submit(new RequestProcessorThread(queue));
31.  
    }
32.  
    }
33.  
     
34.  
    /**
35.  
    * 单例有很多种方式去实现：我采取绝对线程安全的一种方式
36.  
    *
37.  
    * 静态内部类的方式，去初始化单例
38.  
    *
39.  
    * @author Administrator
40.  
    *
41.  
    */
42.  
    private static class Singleton {
43.  
     
44.  
    private static RequestProcessorThreadPool instance;
45.  
     
46.  
    static {
47.  
    instance = new RequestProcessorThreadPool();
48.  
    }
49.  
     
50.  
    public static RequestProcessorThreadPool getInstance() {
51.  
    return instance;
52.  
    }
53.  
     
54.  
    }
55.  
     
56.  
    /**
57.  
    * jvm的机制去保证多线程并发安全
58.  
    *
59.  
    * 内部类的初始化，一定只会发生一次，不管多少个线程并发去初始化
60.  
    *
61.  
    * @return
62.  
    */
63.  
    public static RequestProcessorThreadPool getInstance() {
64.  
    return Singleton.getInstance();
65.  
    }
66.  
     
67.  
    /**
68.  
    * 初始化的便捷方法
69.  
    */
70.  
    public static void init() {
71.  
    getInstance();
72.  
    }
73.  
     
74.  
    }

1.  
   import java.util.ArrayList;
2.  
   import java.util.List;
3.  
   import java.util.Map;
4.  
   import java.util.concurrent.ArrayBlockingQueue;
5.  
   import java.util.concurrent.ConcurrentHashMap;
6.  
    
7.  
   /**
8.  
   * 请求内存队列
9.  
   * @author Administrator
10.  
    *
11.  
    */
12.  
    public class RequestQueue {
13.  
     
14.  
    /**
15.  
    * 内存队列
16.  
    */
17.  
    private List<ArrayBlockingQueue<Request>> queues =
18.  
    new ArrayList<ArrayBlockingQueue<Request>>();
19.  
    /**
20.  
    * 标识位map
21.  
    */
22.  
    private Map<Integer, Boolean> flagMap = new ConcurrentHashMap<Integer, Boolean>();
23.  
     
24.  
    /**
25.  
    * 单例有很多种方式去实现：我采取绝对线程安全的一种方式
26.  
    *
27.  
    * 静态内部类的方式，去初始化单例
28.  
    *
29.  
    * @author Administrator
30.  
    *
31.  
    */
32.  
    private static class Singleton {
33.  
     
34.  
    private static RequestQueue instance;
35.  
     
36.  
    static {
37.  
    instance = new RequestQueue();
38.  
    }
39.  
     
40.  
    public static RequestQueue getInstance() {
41.  
    return instance;
42.  
    }
43.  
     
44.  
    }
45.  
     
46.  
    /**
47.  
    * jvm的机制去保证多线程并发安全
48.  
    *
49.  
    * 内部类的初始化，一定只会发生一次，不管多少个线程并发去初始化
50.  
    *
51.  
    * @return
52.  
    */
53.  
    public static RequestQueue getInstance() {
54.  
    return Singleton.getInstance();
55.  
    }
56.  
     
57.  
    /**
58.  
    * 添加一个内存队列
59.  
    * @param queue
60.  
    */
61.  
    public void addQueue(ArrayBlockingQueue<Request> queue) {
62.  
    this.queues.add(queue);
63.  
    }
64.  
     
65.  
    /**
66.  
    * 获取内存队列的数量
67.  
    * @return
68.  
    */
69.  
    public int queueSize() {
70.  
    return queues.size();
71.  
    }
72.  
     
73.  
    /**
74.  
    * 获取内存队列
75.  
    * @param index
76.  
    * @return
77.  
    */
78.  
    public ArrayBlockingQueue<Request> getQueue(int index) {
79.  
    return queues.get(index);
80.  
    }
81.  
     
82.  
    public Map<Integer, Boolean> getFlagMap() {
83.  
    return flagMap;
84.  
    }
85.  
     
86.  
    }

1.  
   import java.util.concurrent.ArrayBlockingQueue;
2.  
   import java.util.concurrent.Callable;
3.  
    
4.  
   import com.roncoo.eshop.inventory.request.Request;
5.  
    
6.  
   /**
7.  
   * 执行请求的工作线程
8.  
   * @author Administrator
9.  
   *
10.  
    */
11.  
    public class RequestProcessorThread implements Callable<Boolean> {
12.  
     
13.  
    /**
14.  
    * 自己监控的内存队列
15.  
    */
16.  
    private ArrayBlockingQueue<Request> queue;
17.  
     
18.  
    public RequestProcessorThread(ArrayBlockingQueue<Request> queue) {
19.  
    this.queue = queue;
20.  
    }
21.  
     
22.  
    @Override
23.  
    public Boolean call() throws Exception {
24.  
    try {
25.  
    while(true) {
26.  
    // ArrayBlockingQueue
27.  
    // Blocking就是说明，如果队列满了，或者是空的，那么都会在执行操作的时候，阻塞住
28.  
    Request request = queue.take();
29.  
    System.out.println("===========日志===========: 工作线程处理请求，商品id=" + request.getProductId());
30.  
    // 执行这个request操作
31.  
    request.process();
32.  
    }
33.  
    } catch (Exception e) {
34.  
    e.printStackTrace();
35.  
    }
36.  
    return true;
37.  
    }
38.  
     
39.  
    }

2.4.2、两种请求对象封装

1.  
   import com.roncoo.eshop.inventory.model.ProductInventory;
2.  
   import com.roncoo.eshop.inventory.service.ProductInventoryService;
3.  
    
4.  
   /**
5.  
   * 重新加载商品库存的缓存
6.  
   * @author Administrator
7.  
   *
8.  
   */
9.  
   public class ProductInventoryCacheRefreshRequest implements Request {
10.  
     
11.  
    /**
12.  
    * 商品id
13.  
    */
14.  
    private Integer productId;
15.  
    /**
16.  
    * 商品库存Service
17.  
    */
18.  
    private ProductInventoryService productInventoryService;
19.  
     
20.  
    public ProductInventoryCacheRefreshRequest(Integer productId,
21.  
    ProductInventoryService productInventoryService) {
22.  
    this.productId = productId;
23.  
    this.productInventoryService = productInventoryService;
24.  
    }
25.  
     
26.  
    @Override
27.  
    public void process() {
28.  
    // 从数据库中查询最新的商品库存数量
29.  
    ProductInventory productInventory = productInventoryService.findProductInventory(productId);
30.  
    System.out.println("===========日志===========: 已查询到商品最新的库存数量，商品id=" + productId + ", 商品库存数量=" + productInventory.getInventoryCnt());
31.  
    // 将最新的商品库存数量，刷新到redis缓存中去
32.  
    productInventoryService.setProductInventoryCache(productInventory);
33.  
    }
34.  
     
35.  
    public Integer getProductId() {
36.  
    return productId;
37.  
    }
38.  
     
39.  
    }

1.  
   import com.roncoo.eshop.inventory.model.ProductInventory;
2.  
   import com.roncoo.eshop.inventory.service.ProductInventoryService;
3.  
    
4.  
   /**
5.  
   * 比如说一个商品发生了交易，那么就要修改这个商品对应的库存
6.  
   *
7.  
   * 此时就会发送请求过来，要求修改库存，那么这个可能就是所谓的data update request，数据更新请求
8.  
   *
9.  
   * cache aside pattern
10.  
    *
11.  
    * （1）删除缓存
12.  
    * （2）更新数据库
13.  
    *
14.  
    * @author Administrator
15.  
    *
16.  
    */
17.  
    public class ProductInventoryDBUpdateRequest implements Request {
18.  
     
19.  
    /**
20.  
    * 商品库存
21.  
    */
22.  
    private ProductInventory productInventory;
23.  
    /**
24.  
    * 商品库存Service
25.  
    */
26.  
    private ProductInventoryService productInventoryService;
27.  
     
28.  
    public ProductInventoryDBUpdateRequest(ProductInventory productInventory,
29.  
    ProductInventoryService productInventoryService) {
30.  
    this.productInventory = productInventory;
31.  
    this.productInventoryService = productInventoryService;
32.  
    }
33.  
     
34.  
    @Override
35.  
    public void process() {
36.  
    System.out.println("===========日志===========: 数据库更新请求开始执行，商品id=" + productInventory.getProductId() + ", 商品库存数量=" + productInventory.getInventoryCnt());
37.  
    // 删除redis中的缓存
38.  
    productInventoryService.removeProductInventoryCache(productInventory);
39.  
    // 为了模拟演示先删除了redis中的缓存，然后还没更新数据库的时候，读请求过来了，这里可以人工sleep一下
40.  
    // try {
41.  
    // Thread.sleep(20000);
42.  
    // } catch (InterruptedException e) {
43.  
    // e.printStackTrace();
44.  
    // }
45.  
    // 修改数据库中的库存
46.  
    productInventoryService.updateProductInventory(productInventory);
47.  
    }
48.  
     
49.  
    /**
50.  
    * 获取商品id
51.  
    */
52.  
    public Integer getProductId() {
53.  
    return productInventory.getProductId();
54.  
    }
55.  
     
56.  
    }