redis精進 - sortset&延時隊列

最近精進學習Redis，邊學邊寫

先贊後讀，養成習慣

1、SortSet類型使用說明

zset 多是 Redis 提供的最爲特點的數據結構，它也是在面試中面試官最愛問的數據結構。

• 一方面它是set，保證 value 的惟一性，
• 一方面它能夠給每一個 value 一個 score，表明排序權重。

它的內部實現用的是一種叫作「跳躍列表」的數據結構。

2、SortSet經常使用命令

zset 中最後一個 value 被移除後，數據結構自動刪除，內存被回收。

> zadd books 9.0 "think in java"
> zadd books 8.9 "java concurrency"
> zadd books 8.6 "java cookbook"
 > zrange books 0 -1 # 按 score 排序列出，參數區間爲排名範圍
1) "java cookbook"
2) "java concurrency"
3) "think in java"
 > zrevrange books 0 -1 # 按 score 逆序列出，參數區間爲排名範圍
1) "think in java"
2) "java concurrency"
3) "java cookbook"
 > zcard books # 至關於 count()
(integer) 3
 > zscore books "java concurrency" # 獲取指定 value 的 score
"8.9000000000000004"                # 內部 score 使用 double 類型進行存儲，因此存在小數點精度問題
 > zrank books "java concurrency" # 排名
(integer) 1
 > zrangebyscore books 0 8.91 # 根據分值區間遍歷 zset
1) "java cookbook"
2) "java concurrency"
 > zrangebyscore books -inf 8.91 withscores # 根據分值區間 (-∞, 8.91] 遍歷 zset，同時返回分值。inf 表明 infinite，無窮大的意思。
1) "java cookbook"
2) "8.5999999999999996"
3) "java concurrency"
4) "8.9000000000000004"
 > zrem books "java concurrency" # 刪除 value
(integer) 1
> zrange books 0 -1
1) "java cookbook"
2) "think in java"
複製代碼

3、使用場景

排行榜

• 粉絲列表，value 值是粉絲的用戶 ID，score 是關注時間
• 視頻網站須要對用戶上傳的視頻作排行榜，榜單維護多是多方面：按照時間、按照播放量、按照得到的贊數等。

恰巧以前寫的 《Redis實踐熱門文章列表》 就是很好的例子

權重隊列 / 延時隊列

好比：訂單超時未支付，取消訂單，恢復庫存.

對消息隊列有嚴格要求(不能丟)的建議仍是使用kafka,專業的MQ。這些專業的消息中間件提供了不少功能特性，固然他的部署使用維護都是比較麻煩的。若是你對消息隊列沒那麼高要求，想要輕量級的，使用Redis就沒錯啦。

PHP - 延時隊列實例

下面代碼如何加強健壯性：

• function run()須要 lua 來實現器原子性操做
• 能夠考慮 在MySQL，MongoDB等持久數據庫創建 任務隊列 & 已執行隊列 進一步容錯

<?php

trait RedisConnectTrait
{
    private $servers = array();
    private $instances = array();

    /** * 設置 Redis 配置 * @param array $serversConfig * @return $this * @throws Exception */
    private function setServers( array $serversConfig = [ [ '127.0.0.1', 6379, 0.01 ] ] ) {
        if ( !$serversConfig )
            throw new \Exception( 'Redis連接配置不能爲空', false );

        $this->servers = $serversConfig;
        return $this;
    }

    private function initInstances() {
        if (empty($this->instances)) {
            foreach ($this->servers as $server) {
                list($host, $port, $timeout) = $server;

                $redis = new \Redis();
                $redis->connect($host, $port, $timeout);
                // $redis->select( ['index'] );

                $this->instances[] = $redis;
            }
        }
    }
}

class RedisDelayQueueUtil {
    use RedisConnectTrait;

    const QUEUE_PREFIX = 'delay_queue:';
    protected $redis = null;
    protected $key = '';

    public function __construct( string $queueName, array $config = [] ) {
        $instances = $this->setServers( $config )->initInstances();

        $this->key = self::QUEUE_PREFIX . $queueName;
        $this->redis = $instances[ 0 ];
        // $this->redis->auth($config['auth']);
    }

    public function delTask($value) {
        return $this->redis->zRem($this->key, $value);
    }

    public function getTask() {
        //獲取任務，以0和當前時間爲區間，返回一條記錄
        return $this->redis->zRangeByScore( $this->key, 0, time(), [ 'limit' => [ 0, 1 ] ] );
    }

    public function addTask($name, $time, $data) {
        //添加任務，以時間做爲score，對任務隊列按時間從小到大排序
        return $this->redis->zAdd(
            $this->key,
            $time,
            json_encode([
                'task_name' => $name,
                'task_time' => $time,
                'task_params' => $data,
            ], JSON_UNESCAPED_UNICODE )
        );
    }

    public function run() {
        //每次只取一條任務
        $task = $this->getTask();
        if (empty($task)) {
            return false;
        }

        $task = $task[0];
        if ($this->delTask($task)) {
            $task = json_decode($task, true);
            //處理任務
            echo '任務：' . $task['task_name'] . ' 運行時間：' . date('Y-m-d H:i:s') . PHP_EOL;

            return true;
        }

        return false;
    }
}

$dq = new RedisDelayQueueUtil('close_order', [
    'host' => '127.0.0.1',
    'port' => 6379,
    'auth' => '',
    'timeout' => 60,
]);

$dq->addTask('close_order_111', time() + 30, ['order_id' => '111']);
$dq->addTask('close_order_222', time() + 60, ['order_id' => '222']);
$dq->addTask('close_order_333', time() + 90, ['order_id' => '333']);

set_time_limit(0);

$i2Count = 0;
while ( 10 < $i2Count ) {
    $dq->run();
    usleep(100000);
    $i2Count++;
}

複製代碼