PHP中怎么利用Redis 實現消息隊列

本篇文章給大家分享的是有關PHP中怎么利用Redis 實現消息隊列，小編覺得挺實用的，因此分享給大家學習，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

Redis做消息隊列的好處在于它的輕量級，高并發，延遲敏感，應用場景有 即時數據分析、秒殺計數器、緩存等

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

1、消息的可靠性： 沒有相應的機制保證消息的消費，當消費者消費失敗的時候，消息體丟失，需要手動處理。生產者只管向隊列中插入數據，不管消費者是否成功消費。

2、消費者掛掉消息不會丟失，但是需要重新觸發一下消費者，才能夠繼續消費消息。

代碼如下：

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

<?php
/**
 * 獲取數據庫連接
 *
 * @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 是消息發布者，注釋的是將消息存入數據庫部分的代碼。

首先想消息存入 register_users 隊列中，存入的 key是register_users；value是一個list，消息全部存入其中。用 redis-cli 查看數據的命令是：

LRANGE register_users 0 -1

register.php：

<?php
require './lib.php';
$name = $argv[1];
$mobile = $argv[2];
if(empty($name) || empty($mobile)){
    die("參數錯誤");
}
// $connection = getDBConnection('localhost:3306', 'root', 'root', 'blog');
// // 開啟事務
// 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("添加消息隊列失敗");
}
// 發布消息
$redis->publish('register_success', 'ok');
// 所有操作完成后提交事務
// 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("手機號:".$userInfo['mobile']);
                stdout();
                sleep(3);
            }
        }
    }
});

register.php將消息放入redis 的 register_users隊列中，然后再使用 publish 將 register_success 消息發不出去。work.php 使用 subscribe 訂閱register_success 的消息。接收到 register_success 消息之后，讀取 register_users 的消息進行處理。

以上就是PHP中怎么利用Redis 實現消息隊列，小編相信有部分知識點可能是我們日常工作會見到或用到的。希望你能通過這篇文章學到更多知識。更多詳情敬請關注億速云行業資訊頻道。