标签：res ted 通过 fresh boolean ceo ice 没有 end
解决方案总结：
由于数据库层面的读写并发，引发的数据库与缓存数据不一致的问题（本质是后发生的读请求先返回了），可能通过两个小的改动解决：
1）修改服务Service连接池，id取模选取服务连接，能够保证同一个数据的读写都落在同一个后端服务上
“同一个数据的访问一定落到同一个服务上”
获取Service连接的CPool.GetServiceConnection()【返回任何一个可用Service连接】改为CPool.GetServiceConnection(longid)【返回id取模相关联的Service连接】这样的话，就能够保证同一个数据例如uid的请求落到同一个服务Service上
2）修改数据库DB连接池，id取模选取DB连接，能够保证同一个数据的读写在数据库层面是串行的
“让同一数据请求落到同一个数据库链接中“让同一个数据的访问能串行化”
“让同一个数据的访问通过同一条DB连接执行”
“在DB连接池层面稍微修改，按数据取连接即可”
获取DB连接的CPool.GetDBConnection()【返回任何一个可用DB连接】改为CPool.GetDBConnection(longid)【返回id取模相关联的DB连接】
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------
package com.shux.inventory.entity; public class InventoryProduct { private Integer productId; private Long InventoryCnt; public Integer getProductId() { return productId; } public void setProductId(Integer productId) { this.productId = productId; } public Long getInventoryCnt() { return InventoryCnt; } public void setInventoryCnt(Long inventoryCnt) { InventoryCnt = inventoryCnt; } }
public interface Request { public void process(); public Integer getProductId(); public boolean isForceFefresh(); }
package com.shux.inventory.request; import org.springframework.transaction.annotation.Transactional; import com.shux.inventory.biz.InventoryProductBiz; import com.shux.inventory.entity.InventoryProduct; /** ********************************************** * 描述：更新库存信息 * 1、先删除缓存中的数据 * 2、更新数据库中的数据 ********************************************** **/ public class InventoryUpdateDBRequest implements Request{ private InventoryProductBiz inventoryProductBiz; private InventoryProduct inventoryProduct; public InventoryUpdateDBRequest(InventoryProduct inventoryProduct,InventoryProductBiz inventoryProductBiz){ this.inventoryProduct = inventoryProduct; this.inventoryProductBiz = inventoryProductBiz; } @Override @Transactional public void process() { inventoryProductBiz.removeInventoryProductCache(inventoryProduct.getProductId()); inventoryProductBiz.updateInventoryProduct(inventoryProduct); } @Override public Integer getProductId() { // TODO Auto-generated method stub return inventoryProduct.getProductId(); } @Override public boolean isForceFefresh() { // TODO Auto-generated method stub return false; } }
package com.shux.inventory.request; import com.shux.inventory.biz.InventoryProductBiz; import com.shux.inventory.entity.InventoryProduct; /** ********************************************** * 描述：查询缓存数据 * 1、从数据库中查询 * 2、从数据库中查询后插入到缓存中 ********************************************** **/ public class InventoryQueryCacheRequest implements Request { private InventoryProductBiz inventoryProductBiz; private Integer productId; private boolean isForceFefresh; public InventoryQueryCacheRequest(Integer productId,InventoryProductBiz inventoryProductBiz,boolean isForceFefresh) { this.productId = productId; this.inventoryProductBiz = inventoryProductBiz; this.isForceFefresh = isForceFefresh; } @Override public void process() { InventoryProduct inventoryProduct = inventoryProductBiz.loadInventoryProductByProductId(productId); inventoryProductBiz.setInventoryProductCache(inventoryProduct); } @Override public Integer getProductId() { // TODO Auto-generated method stub return productId; } public boolean isForceFefresh() { return isForceFefresh; } public void setForceFefresh(boolean isForceFefresh) { this.isForceFefresh = isForceFefresh; } }
package com.shux.inventory.thread; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import com.shux.inventory.request.Request; import com.shux.inventory.request.RequestQueue; import com.shux.utils.other.SysConfigUtil; /** ********************************************** * 描述：请求处理线程池，初始化队列数及每个队列最多能处理的数量 ********************************************** **/ public class RequestProcessorThreadPool { private static final int blockingQueueNum = SysConfigUtil.get("request.blockingqueue.number")==null?10:Integer.valueOf(SysConfigUtil.get("request.blockingqueue.number").toString()); private static final int queueDataNum = SysConfigUtil.get("request.everyqueue.data.length")==null?100:Integer.valueOf(SysConfigUtil.get("request.everyqueue.data.length").toString()); private ExecutorService threadPool = Executors.newFixedThreadPool(blockingQueueNum); private RequestProcessorThreadPool(){ for(int i=0;i<blockingQueueNum;i++){//初始化队列 ArrayBlockingQueue<Request> queue = new ArrayBlockingQueue<Request>(queueDataNum);//每个队列中放100条数据 RequestQueue.getInstance().addQueue(queue); threadPool.submit(new RequestProcessorThread(queue));//把每个queue交个线程去处理，线程会处理每个queue中的数据 } } public static class Singleton{ private static RequestProcessorThreadPool instance; static{ instance = new RequestProcessorThreadPool(); } public static RequestProcessorThreadPool getInstance(){ return instance; } } public static RequestProcessorThreadPool getInstance(){ return Singleton.getInstance(); } /** * 初始化线程池 */ public static void init(){ getInstance(); } }
package com.shux.inventory.thread; import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.Callable; import com.shux.inventory.request.InventoryUpdateDBRequest; import com.shux.inventory.request.Request; import com.shux.inventory.request.RequestQueue; /** ********************************************** * 描述：请求处理线程 ********************************************** **/ public class RequestProcessorThread implements Callable<Boolean>{ private ArrayBlockingQueue<Request> queue; public RequestProcessorThread(ArrayBlockingQueue<Request> queue){ this.queue = queue; } @Override public Boolean call() throws Exception { Request request = queue.take(); Map<Integer,Boolean> flagMap = RequestQueue.getInstance().getFlagMap(); //不需要强制刷新的时候，查询请求去重处理 if (!request.isForceFefresh()){ if (request instanceof InventoryUpdateDBRequest) {//如果是更新请求，那就置为false flagMap.put(request.getProductId(), true); } else { Boolean flag = flagMap.get(request.getProductId()); /** * 标志位为空，有三种情况 * 1、没有过更新请求 * 2、没有查询请求 * 3、数据库中根本没有数据 * 在最初情况，一旦库存了插入了数据，那就好会在缓存中也会放一份数据， * 但这种情况下有可能由于redis中内存满了，redis通过LRU算法把这个商品给清除了，导致缓存中没有数据 * 所以当标志位为空的时候，需要从数据库重查询一次，并且把标志位置为false，以便后面的请求能够从缓存中取 */ if ( flag == null) { flagMap.put(request.getProductId(), false); } /** * 如果不为空，并且flag为true，说明之前有一次更新请求，说明缓存中没有数据了（更新缓存会先删除缓存）， * 这个时候就要去刷新缓存，即从数据库中查询一次，并把标志位设置为false */ if ( flag != null && flag) { flagMap.put(request.getProductId(), false); } /** * 这种情况说明之前有一个查询请求，并且把数据刷新到了缓存中，所以这时候就不用去刷新缓存了，直接返回就可以了 */ if (flag != null && !flag) { flagMap.put(request.getProductId(), false); return true; } } } request.process(); return true; } }
package com.shux.inventory.request; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ConcurrentHashMap; /** ********************************************** * 描述：请求队列 ********************************************** **/ public class RequestQueue { private List<ArrayBlockingQueue<Request>> queues = new ArrayList<>(); private Map<Integer,Boolean> flagMap = new ConcurrentHashMap<>(); private RequestQueue(){ } private static class Singleton{ private static RequestQueue queue; static{ queue = new RequestQueue(); } public static RequestQueue getInstance() { return queue; } } public static RequestQueue getInstance(){ return Singleton.getInstance(); } public void addQueue(ArrayBlockingQueue<Request> queue) { queues.add(queue); } public int getQueueSize(){ return queues.size(); } public ArrayBlockingQueue<Request> getQueueByIndex(int index) { return queues.get(index); } public Map<Integer,Boolean> getFlagMap() { return this.flagMap; } }
package com.shux.inventory.listener; import org.springframework.context.ApplicationListener; import org.springframework.context.event.ContextRefreshedEvent; import com.shux.inventory.thread.RequestProcessorThreadPool; /** ********************************************** * 描述：spring 启动初始化线程池类 ********************************************** **/ public class InitListener implements ApplicationListener<ContextRefreshedEvent>{ @Override public void onApplicationEvent(ContextRefreshedEvent event) { // TODO Auto-generated method stub if(event.getApplicationContext().getParent() != null){ return; } RequestProcessorThreadPool.init(); } }
package com.shux.inventory.biz; import com.shux.inventory.request.Request; /** ********************************************** * 描述：请求异步处理接口，用于路由队列并把请求加入到队列中 ********************************************** **/ public interface IRequestAsyncProcessBiz { void process(Request request); }
package com.shux.inventory.biz.impl; import java.util.concurrent.ArrayBlockingQueue; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; import com.shux.inventory.biz.IRequestAsyncProcessBiz; import com.shux.inventory.request.Request; import com.shux.inventory.request.RequestQueue; /** ********************************************** * 描述：异步处理请求,用于路由队列并把请求加入到队列中 ********************************************** **/ @Service("requestAsyncProcessService") public class RequestAsyncProcessBizImpl implements IRequestAsyncProcessBiz { private Logger logger = LoggerFactory.getLogger(getClass()); @Override public void process(Request request) { // 做请求的路由，根据productId路由到对应的队列 ArrayBlockingQueue<Request> queue = getQueueByProductId(request.getProductId()); try { queue.put(request); } catch (InterruptedException e) { logger.error("产品ID{}加入队列失败",request.getProductId(),e); } } private ArrayBlockingQueue<Request> getQueueByProductId(Integer productId) { RequestQueue requestQueue = RequestQueue.getInstance(); String key = String.valueOf(productId); int hashcode; int hash = (key == null) ? 0 : (hashcode = key.hashCode())^(hashcode >>> 16); //对hashcode取摸 int index = (requestQueue.getQueueSize()-1) & hash; return requestQueue.getQueueByIndex(index); } }
package com.shux.inventory.controller; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.ResponseBody; import com.shux.inventory.biz.IRequestAsyncProcessBiz; import com.shux.inventory.biz.InventoryProductBiz; import com.shux.inventory.entity.InventoryProduct; import com.shux.inventory.request.InventoryUpdateDBRequest; import com.shux.inventory.request.Request; import com.shux.utils.other.Response; /** ********************************************** * 描述：提交更新请求 ********************************************** **/ @Controller("/inventory") public class InventoryUpdateDBController { private @Autowired InventoryProductBiz inventoryProductBiz; private @Autowired IRequestAsyncProcessBiz requestAsyncProcessBiz; @RequestMapping("/updateDBInventoryProduct") @ResponseBody public Response updateDBInventoryProduct(InventoryProduct inventoryProduct){ Request request = new InventoryUpdateDBRequest(inventoryProduct,inventoryProductBiz); requestAsyncProcessBiz.process(request); return new Response(Response.SUCCESS,"更新成功"); } }
package com.shux.inventory.controller; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.RequestMapping; import com.shux.inventory.biz.IRequestAsyncProcessBiz; import com.shux.inventory.biz.InventoryProductBiz; import com.shux.inventory.entity.InventoryProduct; import com.shux.inventory.request.InventoryQueryCacheRequest; import com.shux.inventory.request.Request; /** ********************************************** * 描述：提交查询请求 * 1、先从缓存中取数据 * 2、如果能从缓存中取到数据，则返回 * 3、如果不能从缓存取到数据，则等待20毫秒，然后再次去数据，直到200毫秒，如果超过200毫秒还不能取到数据，则从数据库中取，并强制刷新缓存数据 ********************************************** **/ @Controller("/inventory") public class InventoryQueryCacheController { private @Autowired InventoryProductBiz inventoryProductBiz; private @Autowired IRequestAsyncProcessBiz requestAsyncProcessBiz; @RequestMapping("/queryInventoryProduct") public InventoryProduct queryInventoryProduct(Integer productId) { Request request = new InventoryQueryCacheRequest(productId,inventoryProductBiz,false); requestAsyncProcessBiz.process(request);//加入到队列中 long startTime = System.currentTimeMillis(); long allTime = 0L; long endTime = 0L; InventoryProduct inventoryProduct = null; while (true) { if (allTime > 200){//如果超过了200ms,那就直接退出，然后从数据库中查询 break; } try { inventoryProduct = inventoryProductBiz.loadInventoryProductCache(productId); if (inventoryProduct != null) { return inventoryProduct; } else { Thread.sleep(20);//如果查询不到就等20毫秒 } endTime = System.currentTimeMillis(); allTime = endTime - startTime; } catch (Exception e) { } } /** * 代码执行到这来，只有以下三种情况 * 1、缓存中本来有数据，由于redis内存满了，redis通过LRU算法清除了缓存，导致数据没有了 * 2、由于之前数据库查询比较慢或者内存太小处理不过来队列中的数据，导致队列里挤压了很多的数据，所以一直没有从数据库中获取数据然后插入到缓存中 * 3、数据库中根本没有这样的数据，这种情况叫数据穿透，一旦别人知道这个商品没有，如果一直执行查询，就会一直查询数据库，如果过多，那么有可能会导致数据库瘫痪 */ inventoryProduct = inventoryProductBiz.loadInventoryProductByProductId(productId); if (inventoryProduct != null) { Request forcRrequest = new InventoryQueryCacheRequest(productId,inventoryProductBiz,true); requestAsyncProcessBiz.process(forcRrequest);//这个时候需要强制刷新数据库，使缓存中有数据 return inventoryProduct; } return null; } }
参考链接：https://www.cnblogs.com/cxxjohnson/p/8519616.html
标签：res ted 通过 fresh boolean ceo ice 没有 end
原文地址：https://www.cnblogs.com/jiawen010/p/12157788.html
