Redis做消息隊列的好處在于它的輕量級，高并發(fā)，延遲敏感。

應(yīng)用場景有即時數(shù)據(jù)分析、秒殺計數(shù)器、緩存等。

Redis做消息隊列待解決的問題：

1.消息的可靠性：

沒有相應(yīng)的機(jī)制保證消息的消費，當(dāng)消費者消費失敗的時候，消息體丟失，需要手動處理。生產(chǎn)者只管向隊列中插入數(shù)據(jù)，不管消費者是否成功消費。

2.消費者掛掉消息不會丟失，但是需要重新觸發(fā)一下消費者，才能夠繼續(xù)消費消息。

代碼如下：

lib.php 是工具文件，里面有數(shù)據(jù)庫的連接、Redis的連接：

<?php /** * 獲取數(shù)據(jù)庫連接 * * @param $host * @param $username * @param $password * @param $database * @return mysqli */function getDBConnection($host, $username, $password, $database){ $connection = new mysqli('p:'.$host, $username, $password, $database); if (!$connection) { echo "Error: Unable to connect to MySQL." . PHP_EOL; echo "Debugging errno: " . mysqli_connect_errno() . PHP_EOL; echo "Debugging error: " . mysqli_connect_error() . PHP_EOL; exit; } mysqli_query($connection, "set names 'utf8'"); return $connection;} /** * 獲取Redis連接 * * @param $host * @param $port * @param string $password * @param int $database * @return Redis */function getRedis($host='127.0.0.1', $port='6379', $password=null, $database=0){ $redis = new Redis(); if(!$redis->connect($host, $port)){ die("Redis連接失敗:IP或端口有誤"); } if(!empty($password) && !$redis->auth($password)){ die("Redis連接失敗:密碼錯誤"); } if($database){ $redis->select($database); } // work中 subscribe 如果一段時間沒有接到消息，就會停掉然后停掉，所以加這個語句讓其永不超時 $redis->setOption(Redis::OPT_READ_TIMEOUT, -1); return $redis;} /** * 打印消息日志 * * @param $msg */function stdout($msg=null){ $msg = '['.date('Y-m-d H:i:s').']'.$msg.chr(10);; fwrite(STDOUT, $msg);}

register.php 是消息發(fā)布者，注釋的是將消息存入數(shù)據(jù)庫部分的代碼。

首先想消息存入 register_users 隊列中，存入的 key是register_users；value是一個list，消息全部存入其中。

用 redis-cli 查看數(shù)據(jù)的命令是：

LRANGE register_users 0 -1

register.php：

<?php require './lib.php';$name = $argv[1];$mobile = $argv[2];if(empty($name) || empty($mobile)){ die("參數(shù)錯誤");}// $connection = getDBConnection('localhost:3306', 'root', 'root', 'blog');// // 開啟事務(wù)// mysqli_begin_transaction($connection);// $sql = "insert into mq_user(name, mobile) values ('$name', '$mobile')";// if(!mysqli_query($connection, $sql)){// die("寫入用戶信息失敗,原因:".$connection->error);// }$redis = getRedis();// 添加消息$result = $redis->lpush('register_users', json_encode(array('name'=>$name, 'mobile'=>$mobile), JSON_UNESCAPED_UNICODE));if($result === false){ mysqli_rollback($connection); die("添加消息隊列失敗");}// 發(fā)布消息$redis->publish('register_success', 'ok');// 所有操作完成后提交事務(wù)// mysqli_commit($connection);// $connection->close();$redis->close();

work.php 做為消息的消費者

<?php require './lib.php';$redis = getRedis();$redis->subscribe(['register_success'], function ($instance, $channelName, $message) { if($channelName == "register_success" && $message = "ok") { $redis = getRedis(); while($redis->lsize("register_users")>0) { $arr = $redis->brPop(['register_users'], 20); if(count($arr)) { $userInfo = json_decode($arr[1], true); stdout("新注冊用戶信息:"); stdout("姓名:".$userInfo['name']); stdout("手機(jī)號:".$userInfo['mobile']); stdout(); sleep(3); } } }});

register.php將消息放入redis 的 register_users隊列中，然后再使用 publish 將 register_success 消息發(fā)不出去。

work.php 使用 subscribe 訂閱register_success 的消息。

接收到 register_success 消息之后，讀取 register_users 的消息進(jìn)行處理。

來源：https://www.cnblogs.com/Lyh1997/p/11491046.html

·END·

PHP開源社區(qū)進(jìn)階·提升·漲薪

總結(jié)

以上是生活随笔為你收集整理的php redis zset 延迟队列_PHP + Redis 实现简单消息队列的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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