需求

  • 异步执行任务
  • 支持定时执行
  • 支持取消任务
  • 保障快速执行

技术背景

  • 基于redis实现
  • php

实现

基于redis的 sorted set + hash，实现定时执行任务的Demo

sorted set 介绍：

  • redis有序集合，且不允许重复的成员，不同的是每个元素都会关联一个double类型的分数
  • redis正是通过分数来为集合中的成员进行从小到大的排序，有序集合的成员是唯一的,但分数(score)却可以重复

思路：

  • 使用sortset类型，将member[成员] =【任务标识】，score[分数] =【定时时间戳】
  • 使用hash类型，将【任务标识】对应的任务数据JSON存到hash中 key =【任务标识】，value =【任务数据JSON】
  • 解决及时消耗，可以运行多个进程进行并行执行

Redis实战之实现定时执行任务 - 图1

php实现代码DEMO

  1. <?php
  2. class DoTest
  3. {
  4.     private const YIELD_KEY = 'yield:list';
  5.     private const YIELD_DATA_KEY = 'yield:data';
  7.     public function run()
  8.     {
  9.         $bbj = RedisClient::instance()->bbj();
  10.         //获取排序（低到高）中第一个task_id
  11.         $data = $bbj->zRange(self::YIELD_KEY, 0, 0, true);
  12.         if (empty($data)) {
  13.             echo "无数据" . PHP_EOL;
  14.             return null;
  15.         }
  16.         $mem = array_keys($data)[0];
  17.         $ts = array_values($data)[0];
  18.         $now = time();
  19.         //校验是否到时
  20.         if ($ts > $now) {
  21.             echo "还未到时间，无需操作" . PHP_EOL;;
  22.             return null;
  23.         }
  24.         //移除集合，多进程并执行到时任务(只能被成功移除一次)
  25.         $row = $bbj->zRem(self::YIELD_KEY, $mem);
  26.         if (empty($row)) {
  27.             echo "已经被剔除" . PHP_EOL;;
  28.             return false;
  29.         }
  30.         //获取当前要执行的任务数据JSON
  31.         $dataJson = $bbj->hGet(self::YIELD_DATA_KEY, $mem);
  32.         //todo 执行定时任务业务逻辑
  33.         var_export($data);
  34.         var_export($dataJson);
  35.         //使用完后删除任务数据JSON
  36.         $bbj->hdel(self::YIELD_DATA_KEY, $mem);
  37.         return true;
  38.     }
  40.     public function add(int $time, $i, string $content)
  41.     {
  42.         $data = [
  43.             'msg' => $content . $time
  44.         ];
  45.         $bbj = RedisClient::instance()->bbj();
  46.         $dataJson = json_encode($data);
  47.         $taskId = $time . '_' . $i;
  48.         $isSc = $bbj->zAdd(self::YIELD_KEY, $time, $taskId);
  49.         if ($isSc) $isSc = $bbj->hSet(self::YIELD_DATA_KEY, $taskId, $dataJson);
  50.         var_export($isSc);
  51.     }
  53.     public function addFeature($i)
  54.     {
  55.         $time = Carbon::now()->addMinute()->timestamp;
  56.         $this->add($time, $i, '未来执行内容');
  57.     }
  59.     public function addCurrent($i)
  60.     {
  61.         $time = time();
  62.         $this->add($time, $i, '马上执行内容');
  63.     }
  65.     /**
  66.      * 取消定时任务，根据任务ID
  67.      * @param $taskId
  68.      */
  69.     public function removeYield($taskId)
  70.     {
  71.         $bbj = RedisClient::instance()->bbj();
  72.         $bbj->zRem(self::YIELD_KEY, $taskId);
  73.         $bbj->hDel(self::YIELD_DATA_KEY, $taskId);
  74.     }
  75. }

监控定时任务队列

<?php
$dt = new DoTest();
while (true) {
    $rt = $dt->run();
    if (is_null($rt)) {
        sleep(1);
    }
}

添加当前执行和未来执行任务

<?php
$dt = new DoTest();
while (true) {
     for ($i = 0; $i < 100000; $i++) {
        // 添加立即执行当前任务
        $dt->addCurrent($i);
        // 添加待执行未来任务
        $dt->addFeature($i);
    }
}

取消定时任务

<?php
$dt = new DoTest();
$dt->removeYield('taskId');

解决场景

  • 定时短信发送/email发送
  • 定时执行??任务