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

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

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

設(shè)計(jì)思路

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

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

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

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

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

實(shí)現(xiàn)步驟

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

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

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

下訂單的方法,下面直接展示代碼，包括請求對象，控制層方法，請求處理線程類的具體實(shí)現(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;//線程安全的請求隊(duì)列
@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;
}
//如果還有剩余商品,就準(zhǔn)備將請求放到請求隊(duì)列中
if(buyqueue==null){//第一次初始化請求隊(duì)列,隊(duì)列的容量為當(dāng)前的商品剩余數(shù)量
buyqueue=new BuyQueue(residue);
}
if(buyqueue.remainingCapacity()>0){//當(dāng)隊(duì)列的可用容量大于0時(shí),將請求放到請求隊(duì)列中
buyqueue.put(buyrequest);
}else{//當(dāng)請求隊(duì)列已滿,本次請求不能處理,直接響應(yīng)客戶端提示請求隊(duì)列已滿
results.put("msg", "搶購隊(duì)列已滿，請稍候重試!");
results.put("done", false);
return results;
}
if(!DealQueueThread.excute){//如果線程類的當(dāng)前執(zhí)行標(biāo)志為未執(zhí)行,即空閑狀態(tài),通過線程池啟動線程
DealQueueThread dealQueue = new DealQueueThread(buyqueue);
ThreadPoolUtil.pool.execute(dealQueue);
BaseLog.info("Thread.activeCount()="+Thread.activeCount());
}
//請求放入到隊(duì)列中，即完成下單請求
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;//線程的默認(rèn)執(zhí)行標(biāo)志為未執(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;//修改線程的默認(rèn)執(zhí)行標(biāo)志為執(zhí)行狀態(tài)
//開始處理請求隊(duì)列中的請求,按照隊(duì)列的FIFO的規(guī)則,先處理先放入到隊(duì)列中的請求
while (buyqueue != null && buyqueue.size() > 0) {
BuyRequest buyreq = buyqueue.take();//取出隊(duì)列中的請求
dealWithQueue(buyreq);//處理請求
}
} catch (InterruptedException e) {
BaseLog.error("DealQueueThread:", e);
} finally {
excute = false;
}
}
public synchronized void dealWithQueue(BuyRequest buyreq) {
try {
//為了盡量確保數(shù)據(jù)的一致性,處理之前先從redis中獲取當(dāng)前搶購商品的剩余數(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ù)量減一,這一步處理可以在隊(duì)列處理完成之后一次性更新剩余數(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)到確認(rèn)訂單并支付的頁面完成支付;

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

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

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