高并发场景下的redis缓存和数据库双写不一致问题分析与解决方案设计

/**注册监听器**/
@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;
  }
}

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();
  }
}

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);
  }
}

  /**
   * 获取商品库存
   */
  @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;
  }