From 74165f8887a0c34ad31cfd518ed73e0562634de6 Mon Sep 17 00:00:00 2001 From: jhx <133451314@qq.com> Date: Fri, 18 Aug 2023 11:36:21 +0800 Subject: [PATCH] uoload --- .gitignore | 2 + composer.json | 12 ++ composer.lock | 21 +++ src/RedisQueue.php | 345 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 380 insertions(+) create mode 100644 .gitignore create mode 100644 composer.json create mode 100644 composer.lock create mode 100644 src/RedisQueue.php diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9fd1dab --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.idea +/vendor/ diff --git a/composer.json b/composer.json new file mode 100644 index 0000000..fd2c791 --- /dev/null +++ b/composer.json @@ -0,0 +1,12 @@ +{ + "name": "woon/stream-queue", + "require": { + "php": ">=7.4", + "ext-redis": "^5.2" + }, + "autoload": { + "psr-4": { + "StreamQueue\\": "src/" + } + } +} diff --git a/composer.lock b/composer.lock new file mode 100644 index 0000000..0631c5e --- /dev/null +++ b/composer.lock @@ -0,0 +1,21 @@ +{ + "_readme": [ + "This file locks the dependencies of your project to a known state", + "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", + "This file is @generated automatically" + ], + "content-hash": "2a270acaeecd3215b4f74fd7a1191ed2", + "packages": [], + "packages-dev": [], + "aliases": [], + "minimum-stability": "stable", + "stability-flags": [], + "prefer-stable": false, + "prefer-lowest": false, + "platform": { + "php": ">=7.4", + "ext-redis": "^5.2" + }, + "platform-dev": [], + "plugin-api-version": "2.1.0" +} diff --git a/src/RedisQueue.php b/src/RedisQueue.php new file mode 100644 index 0000000..7d737cd --- /dev/null +++ b/src/RedisQueue.php @@ -0,0 +1,345 @@ +_mStream = $config['stream']; + + if (is_null($redis)) { + + if (!isset($config['server'])) { + throw new \Exception("you must config the server"); + } + + $tmp = explode(':', $config['server']); + $host = $tmp[0]; + $port = $tmp[1]; + $auth = $tmp[2] ?? null; + + if ($host && $port) { + $this->_mRedis = new Redis(); + $this->_mRedis->connect($host, $port, 1); + if ($auth) { + $this->_mRedis->auth($auth); + } + } else { + throw new \Exception("can not get redis server conf"); + } + } else { + $this->_mRedis = $redis; + } + + if (isset($config['maxLength'])) { + $this->_mMaxLength = $config['maxLength']; + } + + $this->_mGroup = $config['group'] ?? $config['stream']; + $this->_mConsumer = $config['consumer'] ?? $config['stream']; + + $this->creatGroup(); + } + + /* + * 删除当前流(队列) + * */ + public function destoryStream() + { + $this->_mRedis->del($this->_mStream); + } + + /* + * 向流中添加任务 + * $data: array + * $id: string + * */ + public function addTask(array $data, string $id = "*") + { + return $this->_mRedis->xAdd($this->_mStream, $id, $data, $this->_mMaxLength); + } + + /* + * 从group中获取任务 + * $block:阻塞时间，毫秒. null不阻塞 + * $count:读取条数, 只要有数据，条数不够也会立刻返回，即使设置了block + * $start:'>'接受最新数据. 若设置id，则读取大于该id，且未被ack的历史任务 + * + * return [ + * 'id1' => taskdata1, + * 'id2' => taskdata2, + * ... ... + * ] + * + * 无数据返回[] + * */ + public function getTask($block = null, $count = 1, $start = '>') + { + $d = $this->_mRedis->xReadGroup($this->_mGroup, $this->_mConsumer, [$this->_mStream => $start], $count, $block); + + if (is_array($d) && count($d) > 0) { + return $d[$this->_mStream]; + } + + return $d; + } + + /* + * 根据id + * ack任务--从pending中删除 + * 同时从stream中删除 + */ + public function delTask($ids) + { + if (!is_array($ids)) { + $ids = array($ids); + } + $multi = $this->_mRedis->multi(Redis::PIPELINE); + + $multi->xAck($this->_mStream, $this->_mGroup, $ids); + $multi->xDel($this->_mStream, $ids); + $res = $this->_mRedis->exec(); + return $res; + } + + protected function creatGroup($startID = 0) + { + return $this->_mRedis->xGroup('CREATE', $this->_mStream, $this->_mGroup, $startID, true); + } + + /* + * 获取pending队列概要信息 + * return + * + array(4) { + [0] => int(2) //未收到ack的总条数 + [1] => string(15) "1604908088616-0" //startID + [2] => string(15) "1604908088616-1" //endID + [3] => //每个消费者的概要 + array(1) { + [0] => + array(2) { + [0] => string(6) "normal" //消费者名子 + [1] => string(1) "2" //条数 + } + } + } + */ + protected function getPendingInfo() + { + return $this->_mRedis->xPending($this->_mStream, $this->_mGroup); + } + + /* + * 获取pending数据 + * $count: 条数 + * $start: 起始id + * $end: 终止id. - +表示全部 + * $consumer：获取特定consumer的pending数据 + * + * return + *array(1) { + [0] => //单条数据 + array(4) { + [0] => string(15) "1604908973008-0" //id + [1] =>string(6) "normal" //消费者名 + [2] =>int(0) //被取走后经历的毫秒数 + [3] =>int(1) //投递次数 + } + } + */ + protected function getPending($count = 1, $start = '-', $end = '+', $consumer = null) + { + if (!$consumer) { + return $this->_mRedis->xPending($this->_mStream, $this->_mGroup, $start, $end, $count); + } + + return $this->_mRedis->xPending($this->_mStream, $this->_mGroup, $start, $end, $count, $consumer); + } + + /* + * 取[$start, $end]范围内的数据, 注意是闭区间 + * + * $count:条数，null时表示取全部 + * */ + protected function getRange($start = '-', $end = '+', $count = null) + { + if (is_null($count)) { + return $this->_mRedis->xRange($this->_mStream, $start, $end); + } else { + return $this->_mRedis->xRange($this->_mStream, $start, $end, $count); + } + } + + /* + * 将pending队列中超时的数据重新放回队列 + * + * $idleTime: 超时时间, 毫秒 + * $perPage:每次取的任务数 + * + * 注意：只能有一个进行执行pendingRestore + * + * 优点: consumer不需要做任何改动 + * 缺点: + * 先del再add, 成本上不划算， + * 如果del和add中间断掉任务就丢了 + * + * return: restore的数量 + * */ + public function pendingRestore($idleTime = 5000, $perPage = 20) + { + /** + * 比较简单粗暴的取pending数据方式 + * 依赖 + * 1.每次从pending取走/删除超时数据 + * 2.id是按时间排序，小id未超时，大id一定未超时 + * + */ + $restoreNum = 0; + while (1) { + $thisNum = 0; + $data = $this->getPending($perPage); + + foreach ($data as $one) { + $id = $one[0]; + $duration = $one[2]; + if ($duration > $idleTime) { + $data = $this->getRange($id, $id); + $task = $data[$id]; + + $this->delTask($id); + $this->addTask($task); + + $thisNum++; + } + } + + $restoreNum += $thisNum; + + if ($thisNum < $perPage) { + break; + } + } + + return $restoreNum; + } + + /* + * 另一种恢复超时任务的方法 + * 思路：将超时任务放入newConsumer的pending中，后续可以从newConsume的历史中取出数据并处理 + * + * 优点：恢复数据没有重复读，删，插，效率高 + * 缺点： + * consumer需要做改动，至少要改变consumer的名子 + * 只能用单进程从历史数据中读数据，然后处理。 + * + * 详见PendingProcessor类 + * + * $idleTime: 超时时间, 毫秒 + * $newConsumer: 之后处理pending任务的消费者名称 + * $perPage: 每次取pending任务的条数 + * + * return: 满足条件且成功claim的条数 + * */ + public function pendingClaim($idleTime = 5000, $newConsumer = null, $perPage = 20) + { + if (!$newConsumer) { + return false; + } + + $info = $this->getPendingInfo(); + $startID = $info[1]; + $endID = $info[2]; + + $claimNum = 0; + /* + * 使用startid, endid遍历pending列表 + * 因为getpending取的是[startid, endid] + * 所以边界处的id可能被重复取出，但不影响结果的正确性 + * perPage越大/符合xclaim条件的id越多，重复的可能性越小 + * */ + while ($startID != $endID) { + //var_dump([$startID, $endID]); + $data = $this->getPending($perPage, $startID, $endID, $this->_mConsumer); + + foreach ($data as $one) { + $ids[] = $one[0]; + $startID = $one[0]; + } + + $res = $this->_mRedis->xClaim($this->_mStream, $this->_mGroup, $newConsumer, $idleTime, $ids, ['JUSTID']); + + $thisNum = count($res); + $claimNum += $thisNum; + + //id是按时间排列，小id未超时，则后面不会超时 + //在所有id都有相同的投递次数的基础上 + //var_dump($thisNum); + if ($thisNum < $perPage) { + break; + } + } + + return $claimNum; + } +} + +/* demo +$config = [ + 'server' => '10.160.75.237:6379:auth', + 'stream' => 'balltube', + 'consumer' => 'normalprocessor'//可以不设置 + ]; + +//创建队列 +$q = new RedisQueue($config); + +//添加任务 +$task = ['task'=>1]; +$q->addTask($task); + +//获取 +$timeout = 1000; +$task = $q->getTask($timeout); + +//确认并删除 +$taskid = key($task); +$q->delTask($taskid); + +//处理pending +$q->pendingRestore(); +$q->pendingClaim(); + */ \ No newline at end of file