2019獨(dú)角獸企業(yè)重金招聘Python工程師標(biāo)準(zhǔn)>>>

為什么需要消息隊(duì)列

系統(tǒng)中引入消息隊(duì)列機(jī)制是對(duì)系統(tǒng)一個(gè)非常大的改善。例如一個(gè)web系統(tǒng)中，用戶做了某項(xiàng)操作后需要發(fā)送郵件通知到用戶郵箱中。你可以使用同步方式讓用戶等待郵件發(fā)送完成后反饋給用戶，但是這樣可能會(huì)因?yàn)榫W(wǎng)絡(luò)的不確定性造成用戶長(zhǎng)時(shí)間的等待從而影響用戶體驗(yàn)。

有些場(chǎng)景下是不可能使用同步方式等待完成的，那些需要后臺(tái)花費(fèi)大量時(shí)間的操作。例如極端例子，一個(gè)在線編譯系統(tǒng)任務(wù)，后臺(tái)編譯完成需要30分鐘。這種場(chǎng)景的設(shè)計(jì)不可能同步等待后在回饋，必須是先反饋用戶隨后異步處理完成，再等待處理完成后根據(jù)情況再此反饋用戶與否。

另外適用消息隊(duì)列的情況是那些系統(tǒng)處理能力有限的情況下，先使用隊(duì)列機(jī)制把任務(wù)暫時(shí)存放起來(lái)，系統(tǒng)再一個(gè)個(gè)輪流處理掉排隊(duì)的任務(wù)。這樣在系統(tǒng)吞吐量不足的情況下也能穩(wěn)定的處理掉高并發(fā)的任務(wù)。

消息隊(duì)列可以用來(lái)做排隊(duì)機(jī)制，只要系統(tǒng)需要用到排隊(duì)機(jī)制的地方就可以使用消息隊(duì)列來(lái)作。

使用redis怎么做消息隊(duì)列

首先redis它的設(shè)計(jì)是用來(lái)做緩存的，但是由于它自身的某種特性使得他可以用來(lái)做消息隊(duì)列。它有幾個(gè)阻塞式的API可以使用，正是這些阻塞式的API讓他有做消息隊(duì)列的能力。

redis能做消息隊(duì)列得益于他list對(duì)象blpop brpop接口以及Pub/Sub(發(fā)布/訂閱)的某些接口。他們都是阻塞版的，所以可以用來(lái)做消息隊(duì)列。

Redis實(shí)現(xiàn)先進(jìn)先出隊(duì)列

Redis實(shí)現(xiàn)FIFO很容易，只需要一個(gè)List對(duì)象從頭取數(shù)據(jù)，從尾部塞數(shù)據(jù)即可實(shí)現(xiàn)。例如lpush存數(shù)據(jù)，brpop取數(shù)據(jù)。

Redis實(shí)現(xiàn)優(yōu)先級(jí)隊(duì)列

首先brpop和blpop是支持多l(xiāng)ist讀取的，比如brpop lista listb 0 命令就可以實(shí)現(xiàn)先從lista讀取數(shù)據(jù)，讀取完lista的數(shù)據(jù)再去讀取listb的數(shù)據(jù)。

那么我們就可以通過(guò)如下方式實(shí)現(xiàn)了：

127.0.0.1:6379>?lpush?a?1 (integer)?1 127.0.0.1:6379>?lpush?a?2 (integer)?2 127.0.0.1:6379>?lpush?a?3 (integer)?3 127.0.0.1:6379>?lpush?b?1 (integer)?1 127.0.0.1:6379>?lpush?b?2 (integer)?2 127.0.0.1:6379>?lpush?b?3 (integer)?3

127.0.0.1:6379>?brpop?a?b?0 1)?"a" 2)?"1" 127.0.0.1:6379>?brpop?a?b?0 1)?"a" 2)?"2" 127.0.0.1:6379>?brpop?a?b?0 1)?"a" 2)?"3" 127.0.0.1:6379>?brpop?a?b?0 1)?"b" 2)?"1" 127.0.0.1:6379>?brpop?a?b?0 1)?"b" 2)?"2" 127.0.0.1:6379>?brpop?a?b?0 1)?"b" 2)?"3" 127.0.0.1:6379>?brpop?a?b?0

這種方案我們可以支持不同階段的優(yōu)先級(jí)隊(duì)列，例如高中低三個(gè)級(jí)別或者更多的級(jí)別都可以。

?

多優(yōu)先級(jí)問(wèn)題解決

如果優(yōu)先級(jí)級(jí)別很多的情況，假設(shè)有個(gè)這樣的需求，優(yōu)先級(jí)不是簡(jiǎn)單的高中低或者0-10這些固定的級(jí)別。而是類似0-99999這么多級(jí)別。那么我們第三種方案將不太合適了。

雖然redis有sorted set這樣的可以排序的數(shù)據(jù)類型，看是很可惜它沒(méi)有阻塞版的接口。于是我們還是只能使用list類型通過(guò)其他方式來(lái)完成目的。

?

有個(gè)簡(jiǎn)單的做法我們可以只設(shè)置一個(gè)隊(duì)列，并保證它是按照優(yōu)先級(jí)排序號(hào)的。然后通過(guò)二分查找法查找一個(gè)任務(wù)合適的位置，并通過(guò) lset 命令插入到相應(yīng)的位置。?

例如隊(duì)列里面包含著寫優(yōu)先級(jí)的任務(wù)[1, 3, 6, 8, 9, 14]，當(dāng)有個(gè)優(yōu)先級(jí)為7的任務(wù)過(guò)來(lái)，我們通過(guò)自己的二分算法一個(gè)個(gè)從隊(duì)列里面取數(shù)據(jù)出來(lái)反和目標(biāo)數(shù)據(jù)比對(duì)，計(jì)算出相應(yīng)的位置然后插入到指定地點(diǎn)即可。

因?yàn)槎植檎沂潜容^快的，并且redis本身也都在內(nèi)存中，理論上速度是可以保證的。但是如果說(shuō)數(shù)據(jù)量確實(shí)很大的話我們也可以通過(guò)一些方式來(lái)調(diào)優(yōu)。

把上面的方案結(jié)合起來(lái)就會(huì)很大程度上減少開(kāi)銷。例如數(shù)據(jù)量十萬(wàn)的隊(duì)列，它們的優(yōu)先級(jí)也是隨機(jī)0-十萬(wàn)的區(qū)間。我們可以設(shè)置 10個(gè)或者100個(gè)不同的隊(duì)列，0-一萬(wàn)的優(yōu)先級(jí)任務(wù)投放到1號(hào)隊(duì)列，一萬(wàn)-二萬(wàn)的任務(wù)投放到2號(hào)隊(duì)列。這樣將一個(gè)隊(duì)列按不同等級(jí)拆分后它單個(gè)隊(duì)列的數(shù)據(jù) 就減少許多，這樣二分查找匹配的效率也會(huì)高一點(diǎn)。但是數(shù)據(jù)所占的資源基本是不變的，十萬(wàn)數(shù)據(jù)該占多少內(nèi)存還是多少。只是系統(tǒng)里面多了一些隊(duì)列而已。

redis實(shí)現(xiàn)定時(shí)消息隊(duì)列

由于Redis排序集合（Sorted Sets）沒(méi)有實(shí)現(xiàn)阻塞功能，所以只能通過(guò)程序自己實(shí)現(xiàn)。score字段存入時(shí)間戳，由于時(shí)間戳較長(zhǎng)我們用三位數(shù)字代替。

127.0.0.1:6379>?zadd?seta?100?a (integer)?1 127.0.0.1:6379>?zadd?seta?200?b (integer)?1 127.0.0.1:6379>?zadd?seta?300?c (integer)?1 127.0.0.1:6379>?zadd?seta?300?d (integer)?1

首先我們插入4條數(shù)據(jù)。

然后我們獲取0到當(dāng)前時(shí)間的數(shù)據(jù)。比如當(dāng)前時(shí)間戳為200，那么我們執(zhí)行如下命令

127.0.0.1:6379>?zrangebyscore?seta?0?200?limit?0?1 1)?"a" 127.0.0.1:6379>?zrem?seta?a (integer)?1 127.0.0.1:6379>?zrangebyscore?seta?0?201?limit?0?1 1)?"b" 127.0.0.1:6379>?zrem?seta?b (integer)?1 127.0.0.1:6379>?zrangebyscore?seta?0?202?limit?0?1 (empty?list?or?set)

如果取到空數(shù)據(jù)，阻塞一段時(shí)間，然后繼續(xù)取數(shù)據(jù)，循環(huán)執(zhí)行即可。

這里我們?yōu)槭裁礇](méi)有采用zremrangebyscore命令而是采用zrangebyscore和zrem組合，因?yàn)閦remrangebyscore沒(méi)有l(wèi)imit參數(shù)，可能取到多行數(shù)據(jù)（例如兩個(gè)數(shù)據(jù)socore一樣等），由于并發(fā)問(wèn)題可能導(dǎo)致zrem返回0，這樣也沒(méi)事，我們繼續(xù)取即可。

java代碼片段：

public?String?getData()?throws?Exception?{Jedis?jedis?=?getResource();while?(true)?{Set<String>?seta?=?jedis.zrangeByScore("seta",?0,?System.currentTimeMillis(),?0,?1);if?(seta?!=?null?&&?seta.size()?>?0)?{String?data?=?seta.toArray(new?String[]?{})[0];Long?res?=?jedis.zrem("seta",?data);if?(res?>?0)?{return?data;}}Thread.sleep(1000L);} }

來(lái)自個(gè)人博客：http://www.jflyfox.com/mtg/front/article/997.html

轉(zhuǎn)載于:https://my.oschina.net/flyoffox/blog/527860

創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎(jiǎng)勵(lì)來(lái)咯，堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎(jiǎng)

總結(jié)

以上是生活随笔為你收集整理的用redis实现消息队列的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。