[TOC] # 入隊 ~~~ <?php $redis = new Redis(); $redis->connect("127.0.0.1", "6379"); //php客戶端設置的ip及端口 $redis->auth('root'); //密碼 $redis->zAdd('goods:delay:task', time() + 50, json_encode(['id' => 1, 'cid' => 1, 'name' => 'php'])); $redis->zAdd('goods:delay:task', time() + 60, json_encode(['id' => 2, 'cid' => 2, 'name' => 'js'])); $redis->zAdd('goods:delay:task', time() + 70, json_encode(['id' => 3, 'cid' => 3, 'name' => 'py'])); /** * 嘗試3秒內獲取鎖 * @param string $lockName * @param int $timeout * @return bool|string */ function acquireLock($lockName, $timeout = 3) { global $redis; //唯一id $identifier = uniqid(); //當前時間+3秒 $end = time() + $timeout; //當這個時間大于等于當前時間就開始循環 while ($end >= time()) { //設置鎖,key是鎖的名字,值是唯一id,只有鍵key不存在的時候才會設置key的值 //設置成功就返回唯一id if ($redis->set($lockName, $identifier, ['nx'])) { return $identifier; } //暫停1000微秒 usleep(1000); } return false; } /** * 釋放鎖 * @param $lockName * @param $identifier * @return bool */ function releaseLock($lockName, $identifier) { global $redis; while (true) { //監控這個key的變化 $redis->watch($lockName); //如果redis中獲取的這個key等于這個唯一ID if ($redis->get($lockName) == $identifier) { //開啟事務 $redis->multi(Redis::MULTI); //刪除這個key $redis->del($lockName); //執行事務 $res = $redis->exec(); //如果事務返回值是存在,并且等于1,表示事務成功 if (isset($res[0]) && $res[0] == 1) { return true; } } else { //如果獲取的key不等于這個唯一索引 //取消監控 $redis->unwatch(); return false; } } } while (true) { // 因為是有序集合，只要判斷第一條記錄的延時時間，例如第一條未到執行時間 // 相對說明集合的其他任務未到執行時間 //返回有序集 key 中，指定區間內的成員。其中成員的位置按 score 值遞增(從小到大)來排序 $rs = $redis->zRange('goods:delay:task', 0, 0, true);//["{"id":1,"cid":1,"name":"php"}" => 1512713765 ] // 集合沒有任務，睡眠時間設置為5秒 if (empty($rs)) { echo 'no tasks , sleep 5 seconds' . PHP_EOL; sleep(5); continue; } //獲得key $taskJson = key($rs);//{"id":1,"cid":1,"name":"php"} $delay = $rs[$taskJson];//獲得時間 //把存儲的json解析下來 $task = json_decode($taskJson, true); $now = time(); // 到時間執行延時任務 if ($delay <= $now) { // 對當前任務加鎖，避免移動移動延時任務到任務隊列時被其他客戶端修改 //加鎖成功會返回唯一id,不成功會返回false if (!($identifier = acquireLock($task['id']))) { continue; } // 移動延時任務到任務隊列 //刪，移除有序集key中的一個或多個成員，不存在的成員將被忽略 $redis->zRem('goods:delay:task', $taskJson); //增，只能將一個值value插入到列表key的表尾 $redis->rPush('goods:task', $taskJson); echo $task['id'] . ' run ' . PHP_EOL; // 釋放鎖,執行完了開始釋放鎖 releaseLock($task['id'], $identifier); } else { // 延時任務未到執行時間 $sleep = $delay - $now; // 最大值設置為2秒，保證如果有新的任務（延時時間1秒）進入集合時能夠及時的被處理 // $sleep = $sleep > 2 ? 2 :$sleep; echo 'wait ' . $sleep . ' seconds ' . PHP_EOL; sleep($sleep); } } ~~~ # 出隊 ~~~ <?php $redis = new Redis(); $redis->connect("127.0.0.1", "6379"); //php客戶端設置的ip及端口 $redis->auth('root'); //密碼 // 出隊 while (true) { // 阻塞設置超時時間為3秒 //刪除和獲取列表中的第一個元素，或阻塞直到有可用 $task = $redis->blPop(['goods:task'], 3); if ($task) { $redis->rPush('goods:success:task', $task[1]); $task = json_decode($task[1], true); echo $task['id'] . ':' . $task['cid'] . ':' . 'handle success'; echo PHP_EOL; } else { echo 'nothing' . PHP_EOL; sleep(5); } } ~~~