第php基于Redis消息隊列實現(xiàn)的消息推送的方法brpop阻塞模式從隊列右邊獲取值之后刪除

brpoplpush從隊列A的右邊取值之后刪除，從左側(cè)放置到隊列B中

邏輯分析

在普通的任務(wù)腳本中寫入push_queue隊列要發(fā)送消息的目標(biāo)，并為目標(biāo)設(shè)置一個要推送的內(nèi)容，永不過期

RedisPushQueue中brpoplpush處理，處理后的值放到temp_queue，主要防止程序崩潰造成推送失敗

RedisAutoDeleteTempqueueItems處理temp_queue,這里用到了brpop

代碼實現(xiàn)

普通任務(wù)腳本

foreach($user_listas$item){

//命名規(guī)則業(yè)務(wù)類型_操作_ID_隨機6位值自定義我自定義的是"推送內(nèi)容"

$k_name='rabbit_push_'.$item['uid'].'_'.rand(100000,999999);

$redis-lPush('push_queue',$k_name);//左進(jìn)隊列

$redis-set($k_name,'推送內(nèi)容');

RedisPushQueue

//消息隊列處理推送~

//守護(hù)進(jìn)程運行

//nohupphpYOURPATH/RedisPushQueue.php開啟守護(hù)進(jìn)程運行，修改文件之后需要從新啟動

//blpop有值則回去沒值則阻塞主要就是這個函數(shù)在起作用不過并不安全，程序在執(zhí)行過程中崩潰就會導(dǎo)致隊列中的內(nèi)容

//永久丟失~

//BRPOPLPUSH阻塞模式右邊出左邊進(jìn)在填寫隊列內(nèi)容的時候要求從左進(jìn)入

ini_set('default_socket_timeout',-1);//不超時

require_once'YOURPARH/Rongcloud.php';

$redis=new\Redis();

$redis-connect('127.0.0.1',6379);

$redis-select(2);//切換到db2

$redis-setOption(\Redis::OPT_READ_TIMEOUT,-1);

//temp_queue臨時隊列防止程序崩潰導(dǎo)致隊列中內(nèi)容丟失0代表永不超時！

While($key=$redis-brpoplpush('push_queue','temp_queue',0)){

if($val=$redis-get($key)){

//rabbit_push_20_175990

$arr=explode('_',$key);

if(count($arr)!=4){

continue;

$id=$arr[2];

push($id,$val);

//刪除key內(nèi)容

$redis-del($key);

functionpush($id,$v)

//推送操作~

RedisAutoDeleteTempqueueItems

/*自動處理temp_queue中的元素，這個操作是防止RedisPushQueue崩潰的時候做處理

處理思路是使用brpop命令阻塞處理temp_queue這個隊列中的值，如果能獲取到"值"對應(yīng)的"值"，說明RedisPushQueue執(zhí)行失敗了

將值還lpush到push_queue中，以備從新處理

至于為什么使用brpop命令，是因為在RedisPushQueue中我們使用的是brpoplpush

nohupphpYOURPATH/RedisAutoDeleteTempqueueItems.php開啟守護(hù)進(jìn)程運行，修改文件之后需要從新啟動

ini_set('default_socket_timeout',-1);//不超時

$redis=new\Redis();

$redis-connect('127.0.0.1',6379);

$redis-select(2);//切換到db2

$redis-setOption(\Redis::OPT_READ_TIMEOUT,-1);

while($key_arr=$redis-brPop('temp_queue',0)){

if(count($key_arr)!=2){

continue;

$key=$key_arr[1];

if($redis-get($key)){//