Redis隊列實現(xiàn)高并發(fā)的步驟是什么？動力節(jié)點小編來告訴大家。

高并發(fā)的業(yè)務(wù)場景：

我們做商品搶購功能，要面臨的第一個問題就是數(shù)據(jù)不能異常，而保證數(shù)據(jù)不異常我們的解決辦法有很多比如說數(shù)據(jù)庫的鎖機制，或者先改后查的方式都能解決，但是第二個問題來了如果我們用上述技術(shù)解決，數(shù)據(jù)是不會異常了，但是我們的服務(wù)器壓力就會很大了，從而使服務(wù)器宕機，那么我們?nèi)绾问狗?wù)器壓力減小還能保證數(shù)據(jù)不異常呢，我們可以使用隊列的思想，下面我們介紹的是使用redis隊列解決高并發(fā)的問題。

設(shè)計思路

用戶在下訂單之前當然是先查詢到這個商品，在這個查詢的時候，將數(shù)據(jù)庫中商品的剩余數(shù)量存到redis中;

服務(wù)器在一瞬間接到成千上萬的下訂單請求，在控制層沒有直接處理請求數(shù)據(jù)，而是先根據(jù)redis中商品的剩余數(shù)量來判斷，如果>0,就將請求放到請求隊列中，否則直接響應(yīng)客戶端“賣完了”;

考慮到數(shù)據(jù)的一致性，隊列的容量就是商品的剩余數(shù)量，隊列采用的是線程安全的隊列LinkedBlockingQueue(單臺服務(wù)器),然后通過新的線程異步處理這些請求，多臺服務(wù)器的話,可以考慮使用消息隊列MQ,單獨用一臺服務(wù)器去處理消息隊列中的請求;

客戶端發(fā)送訂單請求之后，會收到響應(yīng)，要么是剩余數(shù)量不足(賣完了)，要么是請求已經(jīng)被放到隊列中,為下一步的輪詢訂單做準備;

如果響應(yīng)狀態(tài)是賣完了，直接提示客戶，如果請求已經(jīng)放入隊列中，就可以根據(jù)用戶id和商品id去輪詢訂單了;

實現(xiàn)步驟

說明：用java語言，springmvc框架+redis實現(xiàn)

準備工作，查詢商品信息，將剩余數(shù)量同步到redis中

Jedis jedis = jedisPool.getResource();
BuyGood good=buyGoodService.getById(good_id);
jedis.set("residue"+good_id, good.getResidue()+"");
jedisPool.returnResource(jedis);

下訂單的方法,下面直接展示代碼，包括請求對象，控制層方法，請求處理線程類的具體實現(xiàn)

請求封裝對象

public class BuyRequest {
private int good_id;//商品id
private int user_id;//用戶ID
private int order_id;//訂單id
private BuyOrders buyOrders;//訂單信息
private int response_status;//0:未處理;1:正常;2:異常
public BuyOrders getBuyOrders() {
return buyOrders;
}
public void setBuyOrders(BuyOrders buyOrders) {
this.buyOrders = buyOrders;
}
public int getGood_id() {
return good_id;
}
public void setGood_id(int good_id) {
this.good_id = good_id;
}
public int getOrder_id() {
return order_id;
}
public void setOrder_id(int order_id) {
this.order_id = order_id;
}
public int getResponse_status() {
return response_status;
}
public void setResponse_status(int response_status) {
this.response_status = response_status;
}
public int getUser_id() {
return user_id;
}
public void setUser_id(int user_id) {
this.user_id = user_id;
}
}

處理請求的controller

@Controller
@RequestMapping("/buy")
public class BuyController {
private static BuyQueuebuyqueue =null;//線程安全的請求隊列
@RequestMapping("/addOrders.do")
@ResponseBody
public Object addOrders(BuyRequest buyrequest){
Mapresults = new HashMap<>();
Jedis jedis = jedisPool.getResource();
try {
//下訂單之前，先獲取商品的剩余數(shù)量
int residue = 
Integer.valueOf(jedis.get("residue"+buyrequest.getGood_id()));
if(residue<1){//如果剩余數(shù)量不足，直接響應(yīng)客戶端“賣完了”
results.put("msg", "賣完了");
results.put("done", false);
BaseLog.info("addOrders results="+JSON.toJSONString(results));
return results;
}
//如果還有剩余商品,就準備將請求放到請求隊列中
if(buyqueue==null){//第一次初始化請求隊列,隊列的容量為當前的商品剩余數(shù)量
buyqueue=new BuyQueue(residue);
}
if(buyqueue.remainingCapacity()>0){//當隊列的可用容量大于0時,將請求放到請求隊列中
buyqueue.put(buyrequest);
}else{//當請求隊列已滿,本次請求不能處理,直接響應(yīng)客戶端提示請求隊列已滿
results.put("msg", "搶購隊列已滿，請稍候重試!");
results.put("done", false);
return results;
}
if(!DealQueueThread.excute){//如果線程類的當前執(zhí)行標志為未執(zhí)行,即空閑狀態(tài),通過線程池啟動線程
DealQueueThread dealQueue = new DealQueueThread(buyqueue);
ThreadPoolUtil.pool.execute(dealQueue);
BaseLog.info("Thread.activeCount()="+Thread.activeCount());
}
//請求放入到隊列中，即完成下單請求
results.put("done", true);
results.put("msg", "下訂單成功");
} catch (Exception e) {
results.put("done", false);
results.put("msg", "下單失敗");
BaseLog.info("addOrders results="+JSON.toJSONString(results));
BaseLog.error("addOrders",e);
}finally{
jedisPool.returnResource(jedis);
}
return results;
}
}

處理請求的線程類

@Component
public class DealQueueThread implements Runnable {
private static DealQueueThread dealQueueThread;
@Autowired
BuyGoodService buyGoodService;
@Autowired
BuyOrdersService buyOrdersService;
@Autowired
JedisPool jedisPool;
private Jedis jedis;
private BuyQueuebuyqueue;
public static boolean excute = false;//線程的默認執(zhí)行標志為未執(zhí)行,即空閑狀態(tài)
public DealQueueThread() {
}
public DealQueueThread(BuyQueuebuyqueue) {
this.buyqueue = buyqueue;
jedis = dealQueueThread.jedisPool.getResource();
}
@PostConstruct
public void init() {
dealQueueThread = this;
dealQueueThread.buyGoodService = this.buyGoodService;
dealQueueThread.buyOrdersService = this.buyOrdersService;
dealQueueThread.jedisPool = this.jedisPool;
}
@Override
public void run() {
try {
excute = true;//修改線程的默認執(zhí)行標志為執(zhí)行狀態(tài)
//開始處理請求隊列中的請求,按照隊列的FIFO的規(guī)則,先處理先放入到隊列中的請求
while (buyqueue != null && buyqueue.size() > 0) {
BuyRequest buyreq = buyqueue.take();//取出隊列中的請求
dealWithQueue(buyreq);//處理請求
}
} catch (InterruptedException e) {
BaseLog.error("DealQueueThread:", e);
} finally {
excute = false;
}
}
public synchronized void dealWithQueue(BuyRequest buyreq) {
try {
//為了盡量確保數(shù)據(jù)的一致性,處理之前先從redis中獲取當前搶購商品的剩余數(shù)量
int residue = Integer.valueOf(jedis.get("residue" + 
buyreq.getGood_id()));
if (residue < 1) {//如果沒有剩余商品,就直接返回
buyreq.setResponse_status(3);
return;
}
//如果有剩余商品,先在redis中將剩余數(shù)量減一,再開始下訂單
jedis.decr("residue" + buyreq.getGood_id());
//將數(shù)據(jù)庫中將剩余數(shù)量減一,這一步處理可以在隊列處理完成之后一次性更新剩余數(shù)量
dealQueueThread.buyGoodService.minusResidue(buyreq.getGood_id());
//處理請求,下訂單
BuyOrders bo = new BuyOrders();
bo.setGood_id(buyreq.getGood_id());
bo.setUser_id(buyreq.getUser_id());
int order_id = dealQueueThread.buyOrdersService.insert(bo);
BuyOrders orders = dealQueueThread.buyOrdersService.getById(order_id);
buyreq.setOrder_id(order_id);//訂單id
buyreq.setBuyOrders(orders);//訂單信息
buyreq.setResponse_status(1);//處理完成狀態(tài)
} catch (Exception e) {
buyreq.setResponse_status(2);//異常狀態(tài)
BaseLog.error("DealQueueThread dealWithQueue:", e);
}
}
}

輪詢訂單

思路：查詢訂單和剩余數(shù)量，有以下三種情況：

1.查到訂單，直接跳轉(zhuǎn)到確認訂單并支付的頁面完成支付;

2.還沒有查詢到訂單，但是剩余數(shù)量大于0，說明請求還在隊列中，繼續(xù)輪詢;

3.沒有查到訂單，剩余數(shù)量等于或小于0，說明搶購失敗了，直接響應(yīng)客戶搶購失敗;

以上就是關(guān)于“Redis隊列實現(xiàn)高并發(fā)的步驟”介紹，大家如果想了解更多相關(guān)知識，不妨來關(guān)注一下本站的Redis教程，里面還有更豐富的知識等著大家去學(xué)習(xí)，相信對大家一定會有所幫助的。