Detailed explanation of PHP redis distributed lock and task queue code examples-PHP Tutorial-php.cn

Home

Backend Development

PHP Tutorial

Detailed explanation of PHP redis distributed lock and task queue code examples

伊谢尔伦

Jul 17, 2017 am 09:42 AM

php redis distributed

1. Redis implements distributed lock ideas

The idea is very simple. The main redis function used is setnx(), which should be used to implement distributed locks. The most important function of lock. The first is to store a certain task identification name (here Lock:order is used as an example of the identification name) as a key in redis, and set an expiration time for it. If there is another Lock:order request, first pass setnx() See if Lock:order can be inserted into redis. If it can, it will return true, if not, it will return false. Of course, my code will be more complicated than this idea, and I will explain it further when analyzing the code.

2. Redis implements task queue

The implementation here will use the above Redis distributed lock mechanism, mainly using Redis The data structure of an ordered set. For example, when joining the queue, use the add() function of zset to join the queue, and when leaving the queue, you can use the getScore() function of zset. In addition, several tasks at the top can be popped up.

3. Code analysis

# (1) Let’s first analyze the code implementation of Redis distributed lock　

(1) In order to avoid the lock being unable to be released due to special reasons, after the lock is successfully locked, the lock will be given a survival time (through parameter setting of the lock method or using the default value ). If the survival time is exceeded, The lock will be released automatically. The lock lifetime is short by default (seconds). Therefore, if you need to lock for a long time, you can use the expire method to extend the lock lifetime to an appropriate time, such as in a loop.

(2) System-level locks. When the process crashes for any reason, the operating system will recycle the locks by itself, so there will be no resource loss, but distributed locks are not used. If the one-time setting is very long, Time, once a process crash or other exception occurs due to various reasons and unlock is not called, the lock will become a garbage lock in the remaining time, causing other processes or processes to be unable to enter the locked area after restart.

Let’s look at the locking implementation code first: two main parameters are needed here, one is $timeout, which is the waiting time to acquire the lock cyclically. During this time, it will keep trying to acquire the lock until it times out. If it is 0 , it means returning directly after failing to acquire the lock without waiting; another important parameter is $expire, this parameter refers to the maximum survival time of the current lock, in seconds, it must be greater than 0, if the survival time is exceeded, the lock has not been is released, the system will automatically force the release. Please see the explanation in (1) above for the most important function of this parameter.

Here we first obtain the current time, then obtain the waiting timeout when the lock fails (it is a timestamp), and then obtain the maximum survival time of the lock. The key of redis here uses this format: "Lock: identification name of the lock". The loop begins here. First, insert data into redis, and use the setnx() function. The meaning of this function is, If the key does not exist, insert the data and store the maximum survival time as a value. If the insertion is successful, set the expiration time for the key and place the key in the $lockedName array. Return true, which means the lock is successful. ; If the key exists, the insertion operation will not be performed. There is a rigorous operation here, which is to obtain the remaining time of the current key. If this time is less than 0, it means that there is no survival time set on the key (the key will not exist) , because the previous setnx will automatically create it) If this situation occurs, it is that a certain instance of the process crashes after setnx succeeds, resulting in the subsequent expire not being called. At this time, you can directly set the expire and use the lock for your own use. If the waiting time for lock failure is not set or the maximum waiting time has been exceeded, exit the loop. Otherwise, continue the request after $waitIntervalUs. This is the entire code analysis of locking.

/**
   * 加锁
   * @param [type] $name      锁的标识名
   * @param integer $timeout    循环获取锁的等待超时时间，在此时间内会一直尝试获取锁直到超时，为0表示失败后直接返回不等待
   * @param integer $expire     当前锁的最大生存时间(秒)，必须大于0，如果超过生存时间锁仍未被释放，则系统会自动强制释放
   * @param integer $waitIntervalUs 获取锁失败后挂起再试的时间间隔(微秒)
   * @return [type]         [description]
   */
  public function lock($name, $timeout = 0, $expire = 15, $waitIntervalUs = 100000) {
    if ($name == null) return false;
 
    //取得当前时间
    $now = time();
    //获取锁失败时的等待超时时刻
    $timeoutAt = $now + $timeout;
    //锁的最大生存时刻
    $expireAt = $now + $expire;
 
    $redisKey = "Lock:{$name}";
    while (true) {
      //将rediskey的最大生存时刻存到redis里，过了这个时刻该锁会被自动释放
      $result = $this->redisString->setnx($redisKey, $expireAt);
 
      if ($result != false) {
        //设置key的失效时间
        $this->redisString->expire($redisKey, $expireAt);
        //将锁标志放到lockedNames数组里
        $this->lockedNames[$name] = $expireAt;
        return true;
      }
 
      //以秒为单位，返回给定key的剩余生存时间
      $ttl = $this->redisString->ttl($redisKey);
 
      //ttl小于0 表示key上没有设置生存时间（key是不会不存在的，因为前面setnx会自动创建）
      //如果出现这种状况，那就是进程的某个实例setnx成功后 crash 导致紧跟着的expire没有被调用
      //这时可以直接设置expire并把锁纳为己用
      if ($ttl < 0) {
        $this->redisString->set($redisKey, $expireAt);
        $this->lockedNames[$name] = $expireAt;
        return true;
      }
 
      /*****循环请求锁部分*****/
      //如果没设置锁失败的等待时间 或者 已超过最大等待时间了，那就退出
      if ($timeout <= 0 || $timeoutAt < microtime(true)) break;
 
      //隔 $waitIntervalUs 后继续 请求
      usleep($waitIntervalUs);
 
    }
 
    return false;
  }

Copy after login

Next, let’s look at the unlocking code analysis: Unlocking is much simpler. The incoming parameter is the lock ID. First, determine whether the lock exists. If it exists, delete the lock ID from redis through the deleteKey() function. That’s it.

/**
   * 解锁
   * @param [type] $name [description]
   * @return [type]    [description]
   */
  public function unlock($name) {
    //先判断是否存在此锁
    if ($this->isLocking($name)) {
      //删除锁
      if ($this->redisString->deleteKey("Lock:$name")) {
        //清掉lockedNames里的锁标志
        unset($this->lockedNames[$name]);
        return true;
      }
    }
    return false;
  }
　　　　在贴上删除掉所有锁的方法，其实都一个样，多了个循环遍历而已。
/**
   * 释放当前所有获得的锁
   * @return [type] [description]
   */
  public function unlockAll() {
    //此标志是用来标志是否释放所有锁成功
    $allSuccess = true;
    foreach ($this->lockedNames as $name => $expireAt) {
      if (false === $this->unlock($name)) {
        $allSuccess = false;  
      }
    }
    return $allSuccess;
  }

Copy after login

The above is a summary and sharing of the entire set of ideas and code implementation of distributed locks using Redis. Here I attach the code of an implementation class. In the code, I basically commented each line. , so that everyone can quickly understand and simulate the application. If you want to know more about it, please see the code of the entire class:

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

/**
 *在redis上实现分布式锁
 */
class RedisLock {
  private $redisString;
  private $lockedNames = [];
 
  public function construct($param = NULL) {
    $this->redisString = RedisFactory::get($param)->string;
  }
 
  /**
   * 加锁
   * @param [type] $name      锁的标识名
   * @param integer $timeout    循环获取锁的等待超时时间，在此时间内会一直尝试获取锁直到超时，为0表示失败后直接返回不等待
   * @param integer $expire     当前锁的最大生存时间(秒)，必须大于0，如果超过生存时间锁仍未被释放，则系统会自动强制释放
   * @param integer $waitIntervalUs 获取锁失败后挂起再试的时间间隔(微秒)
   * @return [type]         [description]
   */
  public function lock($name, $timeout = 0, $expire = 15, $waitIntervalUs = 100000) {
    if ($name == null) return false;
 
    //取得当前时间
    $now = time();
    //获取锁失败时的等待超时时刻
    $timeoutAt = $now + $timeout;
    //锁的最大生存时刻
    $expireAt = $now + $expire;
 
    $redisKey = "Lock:{$name}";
    while (true) {
      //将rediskey的最大生存时刻存到redis里，过了这个时刻该锁会被自动释放
      $result = $this->redisString->setnx($redisKey, $expireAt);
 
      if ($result != false) {
        //设置key的失效时间
        $this->redisString->expire($redisKey, $expireAt);
        //将锁标志放到lockedNames数组里
        $this->lockedNames[$name] = $expireAt;
        return true;
      }
 
      //以秒为单位，返回给定key的剩余生存时间
      $ttl = $this->redisString->ttl($redisKey);
 
      //ttl小于0 表示key上没有设置生存时间（key是不会不存在的，因为前面setnx会自动创建）
      //如果出现这种状况，那就是进程的某个实例setnx成功后 crash 导致紧跟着的expire没有被调用
      //这时可以直接设置expire并把锁纳为己用
      if ($ttl < 0) {
        $this->redisString->set($redisKey, $expireAt);
        $this->lockedNames[$name] = $expireAt;
        return true;
      }
 
      /*****循环请求锁部分*****/
      //如果没设置锁失败的等待时间 或者 已超过最大等待时间了，那就退出
      if ($timeout <= 0 || $timeoutAt < microtime(true)) break;
 
      //隔 $waitIntervalUs 后继续 请求
      usleep($waitIntervalUs);
 
    }
 
    return false;
  }
 
  /**
   * 解锁
   * @param [type] $name [description]
   * @return [type]    [description]
   */
  public function unlock($name) {
    //先判断是否存在此锁
    if ($this->isLocking($name)) {
      //删除锁
      if ($this->redisString->deleteKey("Lock:$name")) {
        //清掉lockedNames里的锁标志
        unset($this->lockedNames[$name]);
        return true;
      }
    }
    return false;
  }
 
  /**
   * 释放当前所有获得的锁
   * @return [type] [description]
   */
  public function unlockAll() {
    //此标志是用来标志是否释放所有锁成功
    $allSuccess = true;
    foreach ($this->lockedNames as $name => $expireAt) {
      if (false === $this->unlock($name)) {
        $allSuccess = false;  
      }
    }
    return $allSuccess;
  }
 
  /**
   * 给当前所增加指定生存时间，必须大于0
   * @param [type] $name [description]
   * @return [type]    [description]
   */
  public function expire($name, $expire) {
    //先判断是否存在该锁
    if ($this->isLocking($name)) {
      //所指定的生存时间必须大于0
      $expire = max($expire, 1);
      //增加锁生存时间
      if ($this->redisString->expire("Lock:$name", $expire)) {
        return true;
      }
    }
    return false;
  }
 
  /**
   * 判断当前是否拥有指定名字的所
   * @param [type] $name [description]
   * @return boolean    [description]
   */
  public function isLocking($name) {
    //先看lonkedName[$name]是否存在该锁标志名
    if (isset($this->lockedNames[$name])) {
      //从redis返回该锁的生存时间
      return (string)$this->lockedNames[$name] = (string)$this->redisString->get("Lock:$name");
    }
 
    return false;
  }
 
}

Copy after login

(2) Code analysis of using Redis to implement task queue

(1 ) Task queue, used to put operations that can be processed asynchronously in business logic into the queue, and then dequeue them after being processed in other threads

(2) Distributed locks and other logic are used in the queue to ensure that entry Consistency between enqueue and dequeue

（3）这个队列和普通队列不一样，入队时的id是用来区分重复入队的，队列里面只会有一条记录，同一个id后入的覆盖前入的，而不是追加，如果需求要求重复入队当做不用的任务，请使用不同的id区分

　　先看入队的代码分析：首先当然是对参数的合法性检测，接着就用到上面加锁机制的内容了，就是开始加锁，入队时我这里选择当前时间戳作为score，接着就是入队了，使用的是zset数据结构的add()方法，入队完成后，就对该任务解锁，即完成了一个入队的操作。

/**
   * 入队一个 Task
   * @param [type] $name     队列名称
   * @param [type] $id      任务id（或者其数组）
   * @param integer $timeout    入队超时时间(秒)
   * @param integer $afterInterval [description]
   * @return [type]         [description]
   */
  public function enqueue($name, $id, $timeout = 10, $afterInterval = 0) {
    //合法性检测
    if (empty($name) || empty($id) || $timeout <= 0) return false;
 
    //加锁
    if (!$this->_redis->lock->lock("Queue:{$name}", $timeout)) {
      Logger::get(&#39;queue&#39;)->error("enqueue faild becouse of lock failure: name = $name, id = $id");
      return false;
    }
     
    //入队时以当前时间戳作为 score
    $score = microtime(true) + $afterInterval;
    //入队
    foreach ((array)$id as $item) {
      //先判断下是否已经存在该id了
      if (false === $this->_redis->zset->getScore("Queue:$name", $item)) {
        $this->_redis->zset->add("Queue:$name", $score, $item);
      }
    }
     
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return true;
 
  }

Copy after login

　　接着来看一下出队的代码分析：出队一个Task，需要指定它的$id 和 $score，如果$score与队列中的匹配则出队，否则认为该Task已被重新入队过，当前操作按失败处理。首先和对参数进行合法性检测，接着又用到加锁的功能了，然后及时出队了，先使用getScore()从Redis里获取到该id的score，然后将传入的$score和Redis里存储的score进行对比，如果两者相等就进行出队操作，也就是使用zset里的delete()方法删掉该任务id，最后当前就是解锁了。这就是出队的代码分析。

/**
   * 出队一个Task，需要指定$id 和 $score
   * 如果$score 与队列中的匹配则出队，否则认为该Task已被重新入队过，当前操作按失败处理
   * 
   * @param [type] $name  队列名称 
   * @param [type] $id   任务标识
   * @param [type] $score  任务对应score，从队列中获取任务时会返回一个score，只有$score和队列中的值匹配时Task才会被出队
   * @param integer $timeout 超时时间(秒)
   * @return [type]      Task是否成功，返回false可能是redis操作失败，也有可能是$score与队列中的值不匹配（这表示该Task自从获取到本地之后被其他线程入队过）
   */
  public function dequeue($name, $id, $score, $timeout = 10) {
    //合法性检测
    if (empty($name) || empty($id) || empty($score)) return false;
     
    //加锁
    if (!$this->_redis->lock->lock("Queue:$name", $timeout)) {
      Logger:get(&#39;queue&#39;)->error("dequeue faild becouse of lock lailure:name=$name, id = $id");
      return false;
    }
     
    //出队
    //先取出redis的score
    $serverScore = $this->_redis->zset->getScore("Queue:$name", $id);
    $result = false;
    //先判断传进来的score和redis的score是否是一样
    if ($serverScore == $score) {
      //删掉该$id
      $result = (float)$this->_redis->zset->delete("Queue:$name", $id);
      if ($result == false) {
        Logger::get(&#39;queue&#39;)->error("dequeue faild because of redis delete failure: name =$name, id = $id");
      }
    }
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return $result;
  }

Copy after login

　　学过数据结构这门课的朋友都应该知道，队列操作还有弹出顶部某个值的方法等等，这里处理入队出队操作

/**
   * 获取队列顶部若干个Task 并将其出队
   * @param [type] $name  队列名称
   * @param integer $count  数量
   * @param integer $timeout 超时时间
   * @return [type]      返回数组[0=>[&#39;id&#39;=> , &#39;score&#39;=> ], 1=>[&#39;id&#39;=> , &#39;score&#39;=> ], 2=>[&#39;id&#39;=> , &#39;score&#39;=> ]]
   */
  public function pop($name, $count = 1, $timeout = 10) {
    //合法性检测
    if (empty($name) || $count <= 0) return []; 
     
    //加锁
    if (!$this->_redis->lock->lock("Queue:$name")) {
      Log::get(&#39;queue&#39;)->error("pop faild because of pop failure: name = $name, count = $count");
      return false;
    }
     
    //取出若干的Task
    $result = [];
    $array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
 
    //将其放在$result数组里 并 删除掉redis对应的id
    foreach ($array as $id => $score) {
      $result[] = [&#39;id&#39;=>$id, &#39;score&#39;=>$score];
      $this->_redis->zset->delete("Queue:$name", $id);
    }
 
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return $count == 1 ? (empty($result) ? false : $result[0]) : $result;
  }

Copy after login

　　以上就是用Redis实现任务队列的整一套思路和代码实现的总结和分享

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

/**
 * 任务队列
 * 
 */
class RedisQueue {
  private $_redis;
 
  public function construct($param = null) {
    $this->_redis = RedisFactory::get($param);
  }
 
  /**
   * 入队一个 Task
   * @param [type] $name     队列名称
   * @param [type] $id      任务id（或者其数组）
   * @param integer $timeout    入队超时时间(秒)
   * @param integer $afterInterval [description]
   * @return [type]         [description]
   */
  public function enqueue($name, $id, $timeout = 10, $afterInterval = 0) {
    //合法性检测
    if (empty($name) || empty($id) || $timeout <= 0) return false;
 
    //加锁
    if (!$this->_redis->lock->lock("Queue:{$name}", $timeout)) {
      Logger::get(&#39;queue&#39;)->error("enqueue faild becouse of lock failure: name = $name, id = $id");
      return false;
    }
     
    //入队时以当前时间戳作为 score
    $score = microtime(true) + $afterInterval;
    //入队
    foreach ((array)$id as $item) {
      //先判断下是否已经存在该id了
      if (false === $this->_redis->zset->getScore("Queue:$name", $item)) {
        $this->_redis->zset->add("Queue:$name", $score, $item);
      }
    }
     
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return true;
 
  }
 
  /**
   * 出队一个Task，需要指定$id 和 $score
   * 如果$score 与队列中的匹配则出队，否则认为该Task已被重新入队过，当前操作按失败处理
   * 
   * @param [type] $name  队列名称 
   * @param [type] $id   任务标识
   * @param [type] $score  任务对应score，从队列中获取任务时会返回一个score，只有$score和队列中的值匹配时Task才会被出队
   * @param integer $timeout 超时时间(秒)
   * @return [type]      Task是否成功，返回false可能是redis操作失败，也有可能是$score与队列中的值不匹配（这表示该Task自从获取到本地之后被其他线程入队过）
   */
  public function dequeue($name, $id, $score, $timeout = 10) {
    //合法性检测
    if (empty($name) || empty($id) || empty($score)) return false;
     
    //加锁
    if (!$this->_redis->lock->lock("Queue:$name", $timeout)) {
      Logger:get(&#39;queue&#39;)->error("dequeue faild becouse of lock lailure:name=$name, id = $id");
      return false;
    }
     
    //出队
    //先取出redis的score
    $serverScore = $this->_redis->zset->getScore("Queue:$name", $id);
    $result = false;
    //先判断传进来的score和redis的score是否是一样
    if ($serverScore == $score) {
      //删掉该$id
      $result = (float)$this->_redis->zset->delete("Queue:$name", $id);
      if ($result == false) {
        Logger::get(&#39;queue&#39;)->error("dequeue faild because of redis delete failure: name =$name, id = $id");
      }
    }
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return $result;
  }
 
  /**
   * 获取队列顶部若干个Task 并将其出队
   * @param [type] $name  队列名称
   * @param integer $count  数量
   * @param integer $timeout 超时时间
   * @return [type]      返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
   */
  public function pop($name, $count = 1, $timeout = 10) {
    //合法性检测
    if (empty($name) || $count <= 0) return []; 
     
    //加锁
    if (!$this->_redis->lock->lock("Queue:$name")) {
      Logger::get('queue')->error("pop faild because of pop failure: name = $name, count = $count");
      return false;
    }
     
    //取出若干的Task
    $result = [];
    $array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
 
    //将其放在$result数组里 并 删除掉redis对应的id
    foreach ($array as $id => $score) {
      $result[] = ['id'=>$id, 'score'=>$score];
      $this->_redis->zset->delete("Queue:$name", $id);
    }
 
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return $count == 1 ? (empty($result) ? false : $result[0]) : $result;
  }
 
  /**
   * 获取队列顶部的若干个Task
   * @param [type] $name 队列名称
   * @param integer $count 数量
   * @return [type]     返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
   */
  public function top($name, $count = 1) {
    //合法性检测
    if (empty($name) || $count < 1) return [];
 
    //取错若干个Task
    $result = [];
    $array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
     
    //将Task存放在数组里
    foreach ($array as $id => $score) {
      $result[] = ['id'=>$id, 'score'=>$score];
    }
 
    //返回数组 
    return $count == 1 ? (empty($result) ? false : $result[0]) : $result;    
  }
}

Copy after login

　　到此，这两大块功能基本讲解完毕，对于任务队列，你可以写一个shell脚本，让服务器定时运行某些程序，实现入队出队等操作，这里我就不在将其与实际应用结合起来去实现了，大家理解好这两大功能的实现思路即可，由于代码用的是PHP语言来写的，如果你理解了实现思路，你完全可以使用java或者是.net等等其他语言去实现这两个功能。这两大功能的应用场景十分多，特别是秒杀，另一个就是春运抢火车票，这两个是最鲜明的例子了。当然还有很多地方用到，这里我不再一一列举。

附上分布式锁和任务队列这两个类：

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

/**
 *在redis上实现分布式锁
 */
class RedisLock {
  private $redisString;
  private $lockedNames = [];
 
  public function construct($param = NULL) {
    $this->redisString = RedisFactory::get($param)->string;
  }
 
  /**
   * 加锁
   * @param [type] $name      锁的标识名
   * @param integer $timeout    循环获取锁的等待超时时间，在此时间内会一直尝试获取锁直到超时，为0表示失败后直接返回不等待
   * @param integer $expire     当前锁的最大生存时间(秒)，必须大于0，如果超过生存时间锁仍未被释放，则系统会自动强制释放
   * @param integer $waitIntervalUs 获取锁失败后挂起再试的时间间隔(微秒)
   * @return [type]         [description]
   */
  public function lock($name, $timeout = 0, $expire = 15, $waitIntervalUs = 100000) {
    if ($name == null) return false;
 
    //取得当前时间
    $now = time();
    //获取锁失败时的等待超时时刻
    $timeoutAt = $now + $timeout;
    //锁的最大生存时刻
    $expireAt = $now + $expire;
 
    $redisKey = "Lock:{$name}";
    while (true) {
      //将rediskey的最大生存时刻存到redis里，过了这个时刻该锁会被自动释放
      $result = $this->redisString->setnx($redisKey, $expireAt);
 
      if ($result != false) {
        //设置key的失效时间
        $this->redisString->expire($redisKey, $expireAt);
        //将锁标志放到lockedNames数组里
        $this->lockedNames[$name] = $expireAt;
        return true;
      }
 
      //以秒为单位，返回给定key的剩余生存时间
      $ttl = $this->redisString->ttl($redisKey);
 
      //ttl小于0 表示key上没有设置生存时间（key是不会不存在的，因为前面setnx会自动创建）
      //如果出现这种状况，那就是进程的某个实例setnx成功后 crash 导致紧跟着的expire没有被调用
      //这时可以直接设置expire并把锁纳为己用
      if ($ttl < 0) {
        $this->redisString->set($redisKey, $expireAt);
        $this->lockedNames[$name] = $expireAt;
        return true;
      }
 
      /*****循环请求锁部分*****/
      //如果没设置锁失败的等待时间 或者 已超过最大等待时间了，那就退出
      if ($timeout <= 0 || $timeoutAt < microtime(true)) break;
 
      //隔 $waitIntervalUs 后继续 请求
      usleep($waitIntervalUs);
 
    }
 
    return false;
  }
 
  /**
   * 解锁
   * @param [type] $name [description]
   * @return [type]    [description]
   */
  public function unlock($name) {
    //先判断是否存在此锁
    if ($this->isLocking($name)) {
      //删除锁
      if ($this->redisString->deleteKey("Lock:$name")) {
        //清掉lockedNames里的锁标志
        unset($this->lockedNames[$name]);
        return true;
      }
    }
    return false;
  }
 
  /**
   * 释放当前所有获得的锁
   * @return [type] [description]
   */
  public function unlockAll() {
    //此标志是用来标志是否释放所有锁成功
    $allSuccess = true;
    foreach ($this->lockedNames as $name => $expireAt) {
      if (false === $this->unlock($name)) {
        $allSuccess = false;  
      }
    }
    return $allSuccess;
  }
 
  /**
   * 给当前所增加指定生存时间，必须大于0
   * @param [type] $name [description]
   * @return [type]    [description]
   */
  public function expire($name, $expire) {
    //先判断是否存在该锁
    if ($this->isLocking($name)) {
      //所指定的生存时间必须大于0
      $expire = max($expire, 1);
      //增加锁生存时间
      if ($this->redisString->expire("Lock:$name", $expire)) {
        return true;
      }
    }
    return false;
  }
 
  /**
   * 判断当前是否拥有指定名字的所
   * @param [type] $name [description]
   * @return boolean    [description]
   */
  public function isLocking($name) {
    //先看lonkedName[$name]是否存在该锁标志名
    if (isset($this->lockedNames[$name])) {
      //从redis返回该锁的生存时间
      return (string)$this->lockedNames[$name] = (string)$this->redisString->get("Lock:$name");
    }
 
    return false;
  }
 
}
 
/**
 * 任务队列
 */
class RedisQueue {
  private $_redis;
 
  public function construct($param = null) {
    $this->_redis = RedisFactory::get($param);
  }
 
  /**
   * 入队一个 Task
   * @param [type] $name     队列名称
   * @param [type] $id      任务id（或者其数组）
   * @param integer $timeout    入队超时时间(秒)
   * @param integer $afterInterval [description]
   * @return [type]         [description]
   */
  public function enqueue($name, $id, $timeout = 10, $afterInterval = 0) {
    //合法性检测
    if (empty($name) || empty($id) || $timeout <= 0) return false;
 
    //加锁
    if (!$this->_redis->lock->lock("Queue:{$name}", $timeout)) {
      Logger::get(&#39;queue&#39;)->error("enqueue faild becouse of lock failure: name = $name, id = $id");
      return false;
    }
     
    //入队时以当前时间戳作为 score
    $score = microtime(true) + $afterInterval;
    //入队
    foreach ((array)$id as $item) {
      //先判断下是否已经存在该id了
      if (false === $this->_redis->zset->getScore("Queue:$name", $item)) {
        $this->_redis->zset->add("Queue:$name", $score, $item);
      }
    }
     
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return true;
 
  }
 
  /**
   * 出队一个Task，需要指定$id 和 $score
   * 如果$score 与队列中的匹配则出队，否则认为该Task已被重新入队过，当前操作按失败处理
   * 
   * @param [type] $name  队列名称 
   * @param [type] $id   任务标识
   * @param [type] $score  任务对应score，从队列中获取任务时会返回一个score，只有$score和队列中的值匹配时Task才会被出队
   * @param integer $timeout 超时时间(秒)
   * @return [type]      Task是否成功，返回false可能是redis操作失败，也有可能是$score与队列中的值不匹配（这表示该Task自从获取到本地之后被其他线程入队过）
   */
  public function dequeue($name, $id, $score, $timeout = 10) {
    //合法性检测
    if (empty($name) || empty($id) || empty($score)) return false;
     
    //加锁
    if (!$this->_redis->lock->lock("Queue:$name", $timeout)) {
      Logger:get(&#39;queue&#39;)->error("dequeue faild becouse of lock lailure:name=$name, id = $id");
      return false;
    }
     
    //出队
    //先取出redis的score
    $serverScore = $this->_redis->zset->getScore("Queue:$name", $id);
    $result = false;
    //先判断传进来的score和redis的score是否是一样
    if ($serverScore == $score) {
      //删掉该$id
      $result = (float)$this->_redis->zset->delete("Queue:$name", $id);
      if ($result == false) {
        Logger::get(&#39;queue&#39;)->error("dequeue faild because of redis delete failure: name =$name, id = $id");
      }
    }
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return $result;
  }
 
  /**
   * 获取队列顶部若干个Task 并将其出队
   * @param [type] $name  队列名称
   * @param integer $count  数量
   * @param integer $timeout 超时时间
   * @return [type]      返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
   */
  public function pop($name, $count = 1, $timeout = 10) {
    //合法性检测
    if (empty($name) || $count <= 0) return []; 
     
    //加锁
    if (!$this->_redis->lock->lock("Queue:$name")) {
      Logger::get('queue')->error("pop faild because of pop failure: name = $name, count = $count");
      return false;
    }
     
    //取出若干的Task
    $result = [];
    $array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
 
    //将其放在$result数组里 并 删除掉redis对应的id
    foreach ($array as $id => $score) {
      $result[] = ['id'=>$id, 'score'=>$score];
      $this->_redis->zset->delete("Queue:$name", $id);
    }
 
    //解锁
    $this->_redis->lock->unlock("Queue:$name");
 
    return $count == 1 ? (empty($result) ? false : $result[0]) : $result;
  }
 
  /**
   * 获取队列顶部的若干个Task
   * @param [type] $name 队列名称
   * @param integer $count 数量
   * @return [type]     返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
   */
  public function top($name, $count = 1) {
    //合法性检测
    if (empty($name) || $count < 1) return [];
 
    //取错若干个Task
    $result = [];
    $array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
     
    //将Task存放在数组里
    foreach ($array as $id => $score) {
      $result[] = ['id'=>$id, 'score'=>$score];
    }
 
    //返回数组 
    return $count == 1 ? (empty($result) ? false : $result[0]) : $result;    
  }
}

Copy after login

The above is the detailed content of Detailed explanation of PHP redis distributed lock and task queue code examples. For more information, please follow other related articles on the PHP Chinese website!

Statement of this Website

The content of this article is voluntarily contributed by netizens, and the copyright belongs to the original author. This site does not assume corresponding legal responsibility. If you find any content suspected of plagiarism or infringement, please contact admin@php.cn

Hot AI Tools

Undresser.AI Undress

AI-powered app for creating realistic nude photos

AI Clothes Remover

Online AI tool for removing clothes from photos.

Undress AI Tool

Undress images for free

Clothoff.io

AI clothes remover

Video Face Swap

Swap faces in any video effortlessly with our completely free AI face swap tool!

Hot Article

What's New in Windows 11 KB5054979 & How to Fix Update Issues

4 weeks ago By DDD

How to fix KB5055523 fails to install in Windows 11?

3 weeks ago By DDD

InZoi: How To Apply To School And University

4 weeks ago By DDD

How to fix KB5055518 fails to install in Windows 10?

3 weeks ago By DDD

Where to find the Site Office Key in Atomfall

4 weeks ago By DDD

Hot Tools

Notepad++7.3.1

Easy-to-use and free code editor

SublimeText3 Chinese version

Chinese version, very easy to use

Zend Studio 13.0.1

Powerful PHP integrated development environment

Dreamweaver CS6

Visual web development tools

SublimeText3 Mac version

God-level code editing software (SublimeText3)

Hot Topics

Where is the login entrance for gmail email?

7886

Java Tutorial

1649

CakePHP Tutorial

1410

Laravel Tutorial

1301

PHP Tutorial

1246

Related knowledge

PHP's Purpose: Building Dynamic Websites Apr 15, 2025 am 12:18 AM

PHP is used to build dynamic websites, and its core functions include: 1. Generate dynamic content and generate web pages in real time by connecting with the database; 2. Process user interaction and form submissions, verify inputs and respond to operations; 3. Manage sessions and user authentication to provide a personalized experience; 4. Optimize performance and follow best practices to improve website efficiency and security.

PHP and Python: Different Paradigms Explained Apr 18, 2025 am 12:26 AM

PHP is mainly procedural programming, but also supports object-oriented programming (OOP); Python supports a variety of paradigms, including OOP, functional and procedural programming. PHP is suitable for web development, and Python is suitable for a variety of applications such as data analysis and machine learning.

PHP and Python: Code Examples and Comparison Apr 15, 2025 am 12:07 AM

PHP and Python have their own advantages and disadvantages, and the choice depends on project needs and personal preferences. 1.PHP is suitable for rapid development and maintenance of large-scale web applications. 2. Python dominates the field of data science and machine learning.

Choosing Between PHP and Python: A Guide Apr 18, 2025 am 12:24 AM

PHP is suitable for web development and rapid prototyping, and Python is suitable for data science and machine learning. 1.PHP is used for dynamic web development, with simple syntax and suitable for rapid development. 2. Python has concise syntax, is suitable for multiple fields, and has a strong library ecosystem.

Why Use PHP? Advantages and Benefits Explained Apr 16, 2025 am 12:16 AM

The core benefits of PHP include ease of learning, strong web development support, rich libraries and frameworks, high performance and scalability, cross-platform compatibility, and cost-effectiveness. 1) Easy to learn and use, suitable for beginners; 2) Good integration with web servers and supports multiple databases; 3) Have powerful frameworks such as Laravel; 4) High performance can be achieved through optimization; 5) Support multiple operating systems; 6) Open source to reduce development costs.

PHP: Handling Databases and Server-Side Logic Apr 15, 2025 am 12:15 AM

PHP uses MySQLi and PDO extensions to interact in database operations and server-side logic processing, and processes server-side logic through functions such as session management. 1) Use MySQLi or PDO to connect to the database and execute SQL queries. 2) Handle HTTP requests and user status through session management and other functions. 3) Use transactions to ensure the atomicity of database operations. 4) Prevent SQL injection, use exception handling and closing connections for debugging. 5) Optimize performance through indexing and cache, write highly readable code and perform error handling.

PHP and Python: A Deep Dive into Their History Apr 18, 2025 am 12:25 AM

PHP originated in 1994 and was developed by RasmusLerdorf. It was originally used to track website visitors and gradually evolved into a server-side scripting language and was widely used in web development. Python was developed by Guidovan Rossum in the late 1980s and was first released in 1991. It emphasizes code readability and simplicity, and is suitable for scientific computing, data analysis and other fields.

PHP's Impact: Web Development and Beyond Apr 18, 2025 am 12:10 AM

PHPhassignificantlyimpactedwebdevelopmentandextendsbeyondit.1)ItpowersmajorplatformslikeWordPressandexcelsindatabaseinteractions.2)PHP'sadaptabilityallowsittoscaleforlargeapplicationsusingframeworkslikeLaravel.3)Beyondweb,PHPisusedincommand-linescrip

See all articles