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