上一篇文章我们搭建了一个最基本得springboot+mybatis+redis集群得一个项目。通过这个项目我们来加深理解如何实现缓存与数据库双写一致性。

实现缓存与数据库双写一致性保障方案(一)

大致得思路为:

* 首先要清楚一般是有两种请求:
1:(主动更新数据,数据发生变化,比如库存因为交易成功发送变化)更新数据的时候,根据数据的唯一标识,将操作路由之后,发送到一个jvm内部的队列中

2:读取数据的时候,如果发现数据不在缓存中,那么将重新读取数据+更新缓存的操作,根据唯一标识路由之后,也发送同一个jvm内部的队列中

* 大概方法:将读取数据和更新数据得请求,不要直接去执行这些请求,而且分别放到两条队列中,然后在一条一条得执行。
一个队列对应一个工作线程(比如所有得读取数据得请求放在a队列,更新数据得请求放在b队列)

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

这样的话,就保证了一个数据变更的操作顺序是:先执行,删除缓存,然后再去更新数据库。

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

* 思路优化点:
这里有优化点:

1:一个队列中,其实多个更新缓存(更新缓存是指先从数据库拿到数据,然后再缓存到redis
)请求串在一起是没意义的,因为都是先读数据在缓存到redis中,没必要一直去执行这么多次。因此可以做过滤,如果发现队列中已经有一个更新缓存的请求了,那么就不用再放个更新请求操作进去了,直接等待前面的更新操作请求完成即可

待那个队列对应的工作线程完成了上一个操作的数据库的修改之后,才会去执行下一个操作,也就是缓存更新的操作,此时会从数据库中读取最新的值,然后写入缓存中

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

 

实现思路步骤:

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

* 注册监听器 -- InitListener 实现了线程池和内存队列的初始化 /** * 注册监听器 * @return */
@SuppressWarnings({ "rawtypes", "unchecked" }) @Bean public
ServletListenerRegistrationBean servletListenerRegistrationBean() {
ServletListenerRegistrationBean servletListenerRegistrationBean = new
ServletListenerRegistrationBean();
servletListenerRegistrationBean.setListener(new InitListener()); return
servletListenerRegistrationBean; }
项目启动得时候,就初始化两个队列。这两个队列得生命周期应该跟这个系统得生命周期一样得。

一般我们得java web应用,做系统的初始化,一般在哪里做呢?一般是

ServletContextListener里面做,listener,会跟着整个web应用的启动,就初始化,类似于线程池初始化的构建。

spring boot应用配置监听器是在启动类中(Application),搞一个listener的注册。

* InitListener -- 初始化队列得作用 /** * 系统初始化监听器 * @author Administrator * * 在
Servlet API 中有一个 ServletContextListener 接口, * 它能够监听 ServletContext
对象的生命周期,实际上就是监听 Web 应用的生命周期。 * 当Servlet 容器启动或终止Web 应用时,会触发ServletContextEvent
事件,该事件由ServletContextListener 来处理。 * 在 ServletContextListener
接口中定义了处理ServletContextEvent 事件的两个方法contextInitialized、contextDestroyed * */
public class InitListener implements ServletContextListener { /** * 当Servlet
容器启动Web 应用时调用该方法。在调用完该方法之后,容器再对Filter 初始化, * 并且对那些在Web 应用启动时就需要被初始化的Servlet
进行初始化。 */ @Override public void contextInitialized(ServletContextEvent sce) {
// 初始化工作线程池和内存队列 RequestProcessorThreadPool.init(); } /** * 当Servlet 容器终止Web
应用时调用该方法。在调用该方法之前,容器会先销毁所有的Servlet 和Filter 过滤器。 */ @Override public void
contextDestroyed(ServletContextEvent sce) { } }
自定义监听器实现ServletContextListener接口,在 contextInitialized 初方法中进行始化线程池和内存队列。

* RequestProcessorThreadPool -- 线程池处理类 /** * 请求处理线程池:单例 * @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的机制去保证多线程并发安全 * 因为Singleton 是 静态内部类的方式,保证了RequestProcessorThreadPool只会创建一次。
* 当第二次之后调用就都是拿之前创建好的对象,保证了多线程并发安全。 * 内部类的初始化,一定只会发生一次,不管多少个线程并发去初始化 * * @return
*/ public static RequestProcessorThreadPool getInstance() { return
Singleton.getInstance(); } /** * 初始化的便捷方法 */ public static void init() {
getInstance(); } }
线程池的思路处理,主要是内部使用了一个静态内部类,单例模式初始化线程池和内存队列。

设置线程池大小是多少,每个线程监控的那个内存队列的大小是多少,其实都可以设计成配置的形式,不必要写死,(我这里写死了线程池大小为100,每个线程监听的队列大小为10)
* RequestProcessorThread -- 执行真正请求业务工作的线程类 /** * 执行请求的工作线程 * @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 {
while(true) { //不断消费请求的业务 break; } return true; } }
* RequestQueue -- 请求内存队列 /** * 请求内存队列 * @author Administrator * */ public
class RequestQueue { /** * 内存队列 */ private List<ArrayBlockingQueue<Request>>
queues = new ArrayList<ArrayBlockingQueue<Request>>(); /** *
单例有很多种方式去实现:我采取绝对线程安全的一种方式 * * 静态内部类的方式,去初始化单例 * * @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); } }
初始化生成的所有的队列都会存在这里面。

* Request -- 封装的请求接口,所有的请求方式都会继承它 /** * 请求接口 * @author Administrator * */
public interface Request { void process(); }
 

2、两种请求对象封装

首先要清楚一般是有两种请求:

1:比如发生了交易那么需要更新数据,主动来更新数据这是情况一。更新数据的时候,根据数据的唯一标识,将操作路由之后,发送到一个jvm内部的队列中

2:第二种是读取数据。读取数据的时候,如果发现数据不在缓存中,那么将重新读取数据+更新缓存的操作,根据唯一标识路由之后,也发送同一个jvm内部的队列中

在第一步骤中,我们已经定义好了一个Request接口,目的就是为了所有的不同类型的请求都去实现该接口。因为我们分析了主要有两种请求的可能性(方式)。

所以我们自定义两个请求实现类。

* 封装请求1 -
请求1的设计主要是用来主动更新数据库数据的。因此操作顺序就是先把对应的缓存删掉,然后再去更新具体的数据库数据。请求1中实现了定义了process方法,调用了
service来进行具体的业务操作。 /** * 比如说一个商品发生了交易,那么就要修改这个商品对应的库存 * *
此时就会发送请求过来,要求修改库存,那么这个可能就是所谓的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() { // 删除redis中的缓存
productInventoryService.removeProductInventoryCache(productInventory); //
修改数据库中的库存 productInventoryService.updateProductInventory(productInventory); } }
* 封装请求2 - 请求2的设计主要是用来先去查询库存数据,然后更新到缓存中,所以它
的主要是一个更新缓存的作用的一个请求。请求2中实现了定义了process方法,调用了service来进行具体的业务操作。
业务逻辑就是先查数据库数据,再更新到缓存。 /** * 重新加载商品库存的缓存 * @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); //
将最新的商品库存数量,刷新到redis缓存中去
productInventoryService.setProductInventoryCache(productInventory); } }

在之前封装好的两个请求中,我们已经设计好了service需要的方法,都是设计到具体业务的逻辑。对应的service接口这里就不具体贴出来了,包括mapper也不贴出来看了,因为都是简单的单表操作用来做演示的,相信大家一看就懂怎么取操作哈。现在主要把service接口的实现类贴出来看下,相信大家一看就懂的。四个方法,都是在请求封装类中有用到的。
/** * 商品库存Service实现类 * @author Administrator * */
@Service("productInventoryService") public class ProductInventoryServiceImpl
implements ProductInventoryService { @Resource private ProductInventoryMapper
productInventoryMapper; @Resource private RedisDAO redisDAO; /** * 移除redis中商品缓存
* @param productInventory 商品库存 */ @Override public void
removeProductInventoryCache(ProductInventory productInventory) { String key =
"product:inventory:" + productInventory.getProductId(); redisDAO.delete(key); }
/** * 更新数据库中库存 * @param productInventory 商品库存 */ @Override public void
updateProductInventory(ProductInventory productInventory) {
productInventoryMapper.updateProductInventory(productInventory); } /** *
根据商品id查询商品库存 * @param productId 商品id * @return 商品库存 */ public ProductInventory
findProductInventory(Integer productId) { return
productInventoryMapper.findProductInventory(productId); } /** * 设置redis中商品库存的缓存
* @param productInventory 商品库存 */ public void
setProductInventoryCache(ProductInventory productInventory) { String key =
"product:inventory:" + productInventory.getProductId(); redisDAO.set(key,
String.valueOf(productInventory.getInventoryCnt())); } /** * 获取商品库存的缓存 * @param
productId * @return */ public ProductInventory getProductInventoryCache(Integer
productId) { Long inventoryCnt = 0L; String key = "product:inventory:" +
productId; String result = redisDAO.get(key); if(result != null &&
!"".equals(result)) { try { inventoryCnt = Long.valueOf(result); return new
ProductInventory(productId, inventoryCnt); } catch (Exception e) {
e.printStackTrace(); } } return null; } }
 

3、请求异步执行Service封装

再对以上的涉及到的具体的业务逻辑封装完之后,我们设计一个同意的入口service,进行路由然后根据reque去执行相对的业务操作。

因为前面我们初始化了线程池,并且使用的单例模式,而且还初始化了一个RequestQueue存放了所有的内存队列,并且线程池的数量和每个内存队列的数量,都可以通过我们的配置去进行改变。线程池中的线程监听的自己的一个内存队列,并且这些队列其实也存在我们初始化时定义的那个RequestQueue中,所以我们可以通过操作RequestQueue中的内存队列,就能影响线程中的内存队列去执行对应的操作线程操作。
/** * 请求异步处理的service实现 * @author Administrator * */ public interface
RequestAsyncProcessService { void process(Request request); } @Service public
class RequestAsyncProcessServiceImpl implements RequestAsyncProcessService {
@Override public void process(Request request) { try { //
做请求的路由,根据每个请求的商品id,路由到对应的内存队列中去 ArrayBlockingQueue<Request> queue =
getRoutingQueue(request.getProductId()); //
将请求放入对应的队列中,完成路由操作,put:当Queue已经满了时,会进入等待,只要不被中断,就会插入数据到队列中。会阻塞,可以响应中断
queue.put(request); } catch (Exception e) { e.printStackTrace(); } } /** *
获取路由到的内存队列 * @param productId 商品id * @return 内存队列 */ private
ArrayBlockingQueue<Request> getRoutingQueue(Integer productId) {
//通过RequestQueue获取内存队列。 //因为在初始化线程池的时候,已经调用过一个生成过了,单例模式,所以这里拿到的还是启动时生成的内存队列。
//并且也是线程池中的内存队列,所以当这里拿到队列并且在队列中设置了参数,线程池中的队列就能做出对应操作,因为线程池一直在阻塞中执行。
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; return requestQueue.getQueue(index); } }
 

4、两种请求Controller接口封装
/** * 商品库存Controller * @author Administrator * */ @RestController public class
ProductInventoryController { @Resource private RequestAsyncProcessService
requestAsyncProcessService; @Resource private ProductInventoryService
productInventoryService; /** * 更新商品库存 */
@RequestMapping("/updateProductInventory") @ResponseBody public Response
updateProductInventory(@RequestBody ProductInventory productInventory) {
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; } /** * 获取商品库存 */
@RequestMapping("/getProductInventory") @ResponseBody public ProductInventory
getProductInventory(Integer 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 > 200) { break; } //
尝试去redis中读取一次商品库存的缓存数据 productInventory =
productInventoryService.getProductInventoryCache(productId); // 如果读取到了结果,那么就返回
if(productInventory != null) { return productInventory; } // 如果没有读取到结果,那么等待一段时间
else { Thread.sleep(20); endTime = System.currentTimeMillis(); waitTime =
endTime - startTime; } } // 直接尝试从数据库中读取数据 productInventory =
productInventoryService.findProductInventory(productId); if(productInventory !=
null) { return productInventory; } } catch (Exception e) { e.printStackTrace();
} return new ProductInventory(productId, -1L); } }
到此,就可以进行调用测试了。测试一遍对整个流程可以理解的更清楚。

5、读请求去重优化

情况分析:

对一个商品的库存的数据库更新操作已经在内存队列中了,然后对这个商品的库存的读取操作,要求读取数据库的库存数据,然后更新到缓存中,高并发得情况下,此时有多个读请求过来,这多个读,其实只要有一个读请求操作压到队列里就可以了,其他的读操作,全部都wait那个读请求的操作完后,刷新缓存,就可以读到缓存中的最新数据了,就不用在去到数据库读取了。

 

分析完了具体的情况,我们还应该想到一个问题,因为是多线程高并发的环境,所以我们应该避免高并发而产生的问题,所以这里我们就应该清楚,去重的逻辑代码我们应该写在每个线程的实现类了,避免问题的出现。

这里我们就直接贴出代码来看。
@Override public Boolean call() throws Exception { try { while(true) { //
ArrayBlockingQueue // take和put相互对应,如果队列满了,或者是空的,那么都会在执行操作的时候,阻塞住,一直到能够取到数据
Request request = queue.take(); // 执行这个request操作 boolean forceRfresh =
request.isForceRefresh(); // 先做读请求的去重 if(!forceRfresh) { 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 true; } } }
System.out.println("===========日志===========: 工作线程处理请求,商品id=" +
request.getProductId()); // 执行这个request操作 request.process(); //
假如说,执行完了一个读请求之后,假设数据已经刷新到redis中了 // 但是后面可能redis中的数据会因为内存满了,被自动清理掉 //
如果说数据从redis中被自动清理掉了以后 // 然后后面又来一个读请求,此时如果进来,发现标志位是false,就不会去执行这个刷新的操作了 //
所以在执行完这个读请求之后,实际上这个标志位是停留在false的 } } catch (Exception e) { e.printStackTrace();
} return true; }
*
 在requestQueue中我们定义了唯一的一个map,作为标识符来用。这个标识符,我们在做某一个商品的库存更新的时候,就在这个map中将这个商品标记为true,在缓存刷新的时候,我们就能跟着判断这里面的标识符,来决定是否需要去做缓存刷新。
/** * 标识位map, 做优化操作 * * 我们知道HashMap是线程不安全的,在多线程环境下,使用Hashmap进行put操作会引起死循环, *
导致CPU利用率接近100%,所以在并发情况下不能使用HashMap。 * *
ConcurrentHashMap:主要就是为了应对hashmap在并发环境下不安全而诞生的, *
ConcurrentHashMap的设计与实现非常精巧,大量的利用了volatile,final,CAS等lock-free技术来减少锁竞争对于性能的影响。
*/ private Map<Integer, Boolean> flagMap = new ConcurrentHashMap<Integer,
Boolean>();
*
 通过判断从map中得到的flag,可以通过代码分析。如果flag为true,说明之前已经做过一个更新数据库操作,因此这里标记为false,然后代码会继续去做刷新缓存的操作。当flag为false的时候,说明之前已经做过一个更新数据库操作,并且也刷新过一次缓存了,所以这里就直接返回到controller层了,结合controller层可以知道这里稍微等待一会就会去缓冲中获取数据。然后最后如果flag为null,说明前面什么操作也没做过,所以这里把flag设置为false,去执行一次查询刷新缓存操作,有可能查到的数据是空的就是数据库没有数据。
 

还有一些问题:
1、数据刷入了redis,但是redis LRU算法给清理掉了,标志位还是false
所以此时下一个读请求是从缓存中拿不到数据的,再放一个读Request进队列,让数据去刷新一下
2、可能在200ms内,就是读请求在队列中一直积压着,没有等待到它执行(在实际生产环境中,基本是比较坑了)
所以就直接查一次库,然后给队列里塞进去一个刷新缓存的请求 3、数据库里本身就没有,缓存穿透,穿透redis,请求到达mysql库
解决办法,,在原来的那个controll中做改变。
/** * 获取商品库存 */ @RequestMapping("/getProductInventory") @ResponseBody public
ProductInventory getProductInventory(Integer productId) { ProductInventory
productInventory = null; try { Request request = new
ProductInventoryCacheRefreshRequest( productId, productInventoryService,false);
requestAsyncProcessService.process(request); //
将请求扔给service异步去处理以后,就需要while(true)一会儿,在这里hang住 //
去尝试等待前面有商品库存更新的操作,同时缓存刷新的操作,将最新的数据刷新到缓存中 long startTime =
System.currentTimeMillis(); long endTime = 0L; long waitTime = 0L; //
等待超过200ms没有从缓存中获取到结果 while(true) { if(waitTime > 200) { break; } //
尝试去redis中读取一次商品库存的缓存数据 productInventory =
productInventoryService.getProductInventoryCache(productId); // 如果读取到了结果,那么就返回
if(productInventory != null) { return productInventory; } // 如果没有读取到结果,那么等待一段时间
else { Thread.sleep(20); endTime = System.currentTimeMillis(); waitTime =
endTime - startTime; } } // 直接尝试从数据库中读取数据 productInventory =
productInventoryService.findProductInventory(productId); if(productInventory !=
null) { // 将缓存刷新一下 // 这个过程,实际上是一个读操作的过程,但是没有放在队列中串行去处理,还是有数据不一致的问题 request =
new ProductInventoryCacheRefreshRequest( productId, productInventoryService,
true); requestAsyncProcessService.process(request); // 代码会运行到这里,只有三种情况: //
1、就是说,上一次也是读请求,数据刷入了redis,但是redis LRU算法给清理掉了,标志位还是false //
所以此时下一个读请求是从缓存中拿不到数据的,再放一个读Request进队列,让数据去刷新一下 //
2、可能在200ms内,就是读请求在队列中一直积压着,没有等待到它执行(在实际生产环境中,基本是比较坑了) //
所以就直接查一次库,然后给队列里塞进去一个刷新缓存的请求 // 3、数据库里本身就没有,缓存穿透,穿透redis,请求到达mysql库 return
productInventory; } } catch (Exception e) { e.printStackTrace(); } return new
ProductInventory(productId, -1L); }
 

技术
©2019-2020 Toolsou All rights reserved,
(精华)2020年7月15日 微信小程序 template的使用MySql语句 递归寻找某输入部门的所有下级部门el-select获取选中项label值关于多租户系统的思考python读取、写入txt文本内容(精华)2020年7月13日 微信小程序 页面间通信(精华)2020年6月26日 C#类库 异常处理帮助类SpringBoot JpaRepository 数据库增删改查在vue+element-ui中,select选项值动态从后台获取,同时将选中值的id传给后台的方法冲突声明(conflicting declaration)解决