建站
咨詢
售后
建站咨詢
新聞中心
[[170879]]
創(chuàng)新互聯(lián)建站從2013年成立，是專業(yè)互聯(lián)網(wǎng)技術(shù)服務(wù)公司，擁有項目成都網(wǎng)站建設(shè)、網(wǎng)站制作網(wǎng)站策劃，項目實施與項目整合能力。我們以讓每一個夢想脫穎而出為使命，1280元隆林做網(wǎng)站,已為上家服務(wù),為隆林各地企業(yè)和個人服務(wù),聯(lián)系電話:18982081108
隊列在數(shù)據(jù)結(jié)構(gòu)中是一種線性表，從一端插入數(shù)據(jù)，然后從另一端刪除數(shù)據(jù)。本文目的不是講解各種隊列算法，而是在應(yīng)用層面講述使用隊列能解決哪些場景問題。
在我開發(fā)過的系統(tǒng)中，不是所有的業(yè)務(wù)都必須實時處理、不是所有的請求都必須實時反饋結(jié)果給用戶、不是所有的請求/處理都必須100%處理成功、不知道誰依賴“我”的處理結(jié)果、不關(guān)心其他系統(tǒng)如何處理后續(xù)業(yè)務(wù)、不需要強一致性，只需保證最終一致性即可、想要保證數(shù)據(jù)處理的有序性;此時你應(yīng)該考慮使用隊列來解決這些問題。在實際開發(fā)時我們經(jīng)常使用隊列進行異步處理、系統(tǒng)解耦、數(shù)據(jù)同步、流量削峰、緩沖、限流等。
應(yīng)用場景
異步處理：使用隊列的一個主要原因是進行異步處理，比如用戶注冊成功后需要發(fā)送注冊成功郵件/新用戶積分/優(yōu)惠券等等、緩存過期時先返回老的數(shù)據(jù)，然后異步更新緩存、異步寫日志等;通過異步處理，可以提升主流程響應(yīng)速度，而非主流程/非重要業(yè)務(wù)可以異步集中處理，這樣還可以將任務(wù)聚合然后批量處理;因此可以使用消息隊列/任務(wù)隊列來進行異步處理。
系統(tǒng)解耦：比如用戶成功支付完成訂單后，需要通知生產(chǎn)配貨系統(tǒng)、發(fā)票系統(tǒng)、庫存系統(tǒng)、推薦系統(tǒng)、搜索系統(tǒng)、風(fēng)控系統(tǒng)等進行業(yè)務(wù)處理;而未來需要添加/支持哪些業(yè)務(wù)是不清楚的，而且這些業(yè)務(wù)處理不需要實時處理、不需要強一致，只需要最終一致性即可，因此可以通過消息隊列/任務(wù)隊列進行系統(tǒng)解耦。
數(shù)據(jù)同步：比如想把Mysql變更的數(shù)據(jù)同步到Redis、或者將Mysql數(shù)據(jù)同步到Mongodb、或者機房間數(shù)據(jù)同步、或者主從數(shù)據(jù)同步等，此時可以考慮使用如databus、canal、otter。使用數(shù)據(jù)總線隊列進行數(shù)據(jù)同步的好處是可以保證數(shù)據(jù)修改的有序性。
流量削峰：系統(tǒng)瓶頸一般在數(shù)據(jù)庫上，比如扣減庫存、下單等;此時可以考慮使用隊列將變更請求暫時放入隊列，通過緩存+隊列暫存的方式將數(shù)據(jù)庫流量削峰;還有如秒殺系統(tǒng)，下單服務(wù)會是該系統(tǒng)的瓶頸，此時會使用隊列進行排隊和限流，從而保護下單服務(wù)。通過隊列暫存或者隊列限流來削峰。
比如減庫存，可以考慮這樣設(shè)計：
直接在Redis中扣減，然后記錄下扣減日志(FIFO隊列)，通過Worker去同步到DB。
實際隊列的應(yīng)用場景還是非常多的，本文列舉了筆者遇到過比較多的場景。
緩沖區(qū)隊列
典型的如Log4j的日志緩沖區(qū)，當(dāng)我們使用log4j記錄日志時，可以配置字節(jié)緩沖區(qū)，字節(jié)緩存區(qū)滿時會立即同步到磁盤(flush操作)。Log4j使用BufferedWriter實現(xiàn)的;此模式不是異步寫，在緩沖區(qū)滿的時候還是會阻塞主線程。如果需要異步模式可以使用AsyncAppender，然后通過bufferSize控制日志事件緩沖區(qū)大小。
通過緩沖區(qū)隊列可以實現(xiàn)：批量處理、異步處理。
任務(wù)隊列
使用任務(wù)隊列將一些不需要與主線程同步執(zhí)行的任務(wù)扔到任務(wù)隊列異步處理即可;筆者用的最多的是線程池任務(wù)隊列(默認(rèn)LinkedBlockingQueue)和Disruptor任務(wù)隊列(RingBuffer)。如刷數(shù)據(jù)時，將任務(wù)扔到隊列異步處理即可，處理成功后再異步通知用戶;還有如刪除SKU操作，用戶請求時直接將任務(wù)分解并扔到隊列，異步處理，處理成功后異步通知用戶即可;還有如查詢聚合，將多個可并行處理的任務(wù)扔到隊列然后等待最慢的一個返回。如果使用的是內(nèi)存任務(wù)隊列請記住可能存在系統(tǒng)重啟等問題造成的數(shù)據(jù)丟失。
通過任務(wù)隊列可以實現(xiàn)：異步處理、任務(wù)分解/聚合處理。
注：JDK7提供了ExecutorService的新的實現(xiàn)ForkJoinPool，其提供了Work-stealing機制，可以更好地提升并發(fā)效率。
在使用Executors.newFixedThreadPool時，其沒有設(shè)置隊列大小(默認(rèn)Integer.MAX_VALUE)，如果有大量任務(wù)被緩存到LinkedBlockingQueue中等待線程執(zhí)行，會出現(xiàn)GC慢等問題，造成系統(tǒng)響應(yīng)慢甚至OOM。因此在使用線程池時候，要指定隊列大小并設(shè)置合理的RejectedExecutionHandler;要記錄請求來源的參數(shù)方便定位引發(fā)問題的源頭。
消息隊列
筆者所在公司使用的是自研的JMQ;開源的有ActiveMQ、Kafka、Redis。使用消息隊列存儲各業(yè)務(wù)數(shù)據(jù)，其他系統(tǒng)根據(jù)需要訂閱即可。常見的模式是：點對點(一個消息只有一個消費者)、發(fā)布訂閱(一個消息可以有多個消費者);而常用的是發(fā)布訂閱模式。
比如用戶注冊成功、修改商品數(shù)據(jù)、訂單狀態(tài)變更等都應(yīng)該將變更發(fā)送到消息隊列，從而其他系統(tǒng)根據(jù)需要訂閱該消息，然后按照自己的需求進行業(yè)務(wù)邏輯開發(fā)。
在添加新功能時，消息消費者只需要訂閱該消息，然后開發(fā)相應(yīng)的業(yè)務(wù)邏輯，消息生產(chǎn)者根本不關(guān)心你怎么使用消息和你做什么業(yè)務(wù)處理。
同步調(diào)用，添加什么新功能都需要到用戶系統(tǒng)提需求。其中一個服務(wù)出現(xiàn)問題了，整個服務(wù)就不可用了。
消息隊列，用戶系統(tǒng)只需要發(fā)布用戶注冊成功的消息即可，相關(guān)系統(tǒng)訂閱該消息，然后執(zhí)行相關(guān)的業(yè)務(wù)邏輯。相關(guān)服務(wù)出問題不影響到注冊主流程。
通過消息隊列可以實現(xiàn)：異步處理、系統(tǒng)解耦。
請求隊列
請求隊列是指如在Web環(huán)境下對用戶請求排隊，從而進行一些特殊控制：流量控制、請求分級、請求隔離;如將請求按照功能劃分到不同的隊列，從而使得不同的隊列出現(xiàn)問題后相互不影響;還可以對請求分級，一些重要請求可以優(yōu)先處理(發(fā)展到一定程度應(yīng)將功能物理分離);還有服務(wù)器處理能力有限，在接近服務(wù)器瓶頸時需要考慮限流，最簡單的限流時丟棄處理不了的請求，此時可以使用隊列進行流量控制。
數(shù)據(jù)總線隊列
一般消息隊列中的消息都是業(yè)務(wù)維度的，比如業(yè)務(wù)鍵或者業(yè)務(wù)狀態(tài)等，比如哪個SKU變更了，而有些訂閱者需要再查一遍來獲取***的修改數(shù)據(jù)(比如緩存同步);通過現(xiàn)有的消息隊列方式的缺點是很難只進行修改部分的推送和保證數(shù)據(jù)有序性。而此種場景比較適合使用數(shù)據(jù)總線隊列實現(xiàn)。如數(shù)據(jù)庫數(shù)據(jù)修改后需要同步數(shù)據(jù)到緩存，或者需要將一個機房數(shù)據(jù)同步到另一個機房，只是數(shù)據(jù)維度的同步，此時應(yīng)該使用數(shù)據(jù)總線隊列如canal、otter、databus;使用數(shù)據(jù)總線隊列的好處是可以保證數(shù)據(jù)的有序性。
混合隊列
在《構(gòu)建需求響應(yīng)式億級商品詳情頁》曾介紹過該方式的隊列，使用混合隊列來解決實際問題。
此處MQ是使用京東自研的JMQ，消息是可靠持久化存儲的;應(yīng)用會按照不同的維度發(fā)布消息到JMQ;下游應(yīng)用接收到該消息后會放入到Redis，使用Redis List來存儲這些任務(wù);應(yīng)用將Redis消息消費處理后，會按照不同的維度聚合商品消息然后再次發(fā)送出去。
使用Redis隊列的主要原因是想提升消息堆積能力和并發(fā)處理能力。另外在使用Redis構(gòu)建消息隊列時需要考慮網(wǎng)絡(luò)抖動造成的消息丟失問題，因為Redis是沒有回滾事務(wù)的，或者說是確認(rèn)機制。我們使用如下方式防止消息丟失：
- try {
- id = queueRedis.opsForList().rightPopAndLeftPush(queueName, processingQueueName);
- } catch (Exception e) {
//發(fā)生了網(wǎng)絡(luò)異常，需要把processing中的id再放回到waiting queue中
- String msg = queueName + " to " + processingQueueName + " rpoplpush error";
- LOG.error(msg, e);
- //報警代碼
- }
而對于失敗我們會進行重試三次，重試失敗后放入失敗隊列，而失敗隊列是具有防重功能的(從本地隊列和失敗隊列排重)，使用的是Redis Lua腳本實現(xiàn)：
- static EventQueueScript ADD_TO_FAIL_QUEUE_REDIS_SCRIPT = new EventQueueScript(
- "redis.call('lrem', KEYS[1], 1, ARGV[1]) redis.call('lrem', KEYS[2], 1, ARGV[1]) return redis.call('lpush', KEYS[2], ARGV[1])"
- );
Redis作者Antirez開發(fā)的內(nèi)存分布式消息隊列Disque是未來更好的內(nèi)存消息隊列選擇。
其他
優(yōu)先級隊列：在實際開發(fā)時肯定有些任務(wù)是緊急的，此時應(yīng)該優(yōu)先處理緊急的任務(wù);所以請考慮對隊列進行分級。
副本隊列：在進行一些系統(tǒng)重構(gòu)或者上新的功能時，如果沒有足夠的信心保證業(yè)務(wù)邏輯正確，可以考慮存儲一份隊列的副本(比如1小時、1天的)，從而當(dāng)業(yè)務(wù)出現(xiàn)問題時可以對這些消息進行回放。
鏡像隊列：每個隊列不會***制訂閱數(shù)量，一定會有一個極限的;當(dāng)?shù)竭_極限時請考慮使用鏡像隊列方式解決該問題。
隊列并發(fā)數(shù)：不同隊列實現(xiàn)，隊列服務(wù)端并發(fā)連接數(shù)是不一樣的;一定不是增大隊列并發(fā)連接數(shù)消費能力也隨著增加;也不會因為增加了消費服務(wù)器消費并發(fā)能力也隨著增加，需要根據(jù)實際情況來設(shè)置合理的并發(fā)連接數(shù)。
推還是拉：消息體內(nèi)容不是越全越好，需要根據(jù)具體業(yè)務(wù)設(shè)計消息體;如有些系統(tǒng)依賴商品變更消息(只有一個SKU)、有些系統(tǒng)依賴商品狀態(tài)消息(SKU、狀態(tài))、有些系統(tǒng)依賴商品屬性變更消息(SKU、變更的屬性)等，如果讓所有系統(tǒng)都消費商品變更消息，那么這些系統(tǒng)都會調(diào)用商品查詢服務(wù)拉一下***的商品信息然后進行處理。因此要根據(jù)實際情況來決定是使用推送方式(將系統(tǒng)需要的所有信息推過去)還是拉取方式(只推送ID，然后再查一遍)。
消息合并：如果消息寫入量非常大，應(yīng)該考慮將消息合并寫，可以"寫應(yīng)用本地磁盤隊列"-->“同步本地磁盤隊列到消息中間件”;同步時可以根據(jù)需求制定同步策略，如1秒同步1次。
新聞名稱：高并發(fā)系統(tǒng)中隊列術(shù)的哪些應(yīng)用場景
本文網(wǎng)址：http://www.fisionsoft.com.cn/article/dhdshpe.html
