1. 新增error max count

2. 新增本地error表
workbunny · Feb 26, 2024 · f37a363 · f37a363
1 parent 0ca496d
commit f37a363
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 6 deletions.
diff --git a/src/Builders/Traits/MessageQueueMethod.php b/src/Builders/Traits/MessageQueueMethod.php
@@ -7,6 +7,7 @@
 use Workbunny\WebmanRqueue\Exceptions\WebmanRqueueException;
 use Workbunny\WebmanRqueue\Headers;
 use Workerman\Worker;
+use function Workbunny\WebmanRqueue\config;
 
 trait MessageQueueMethod
 {
@@ -426,12 +427,27 @@ public function consume(Worker $worker, bool $del = true): bool
                             $header->_error = "{$throwable->getMessage()} [{$throwable->getFile()}:{$throwable->getLine()}]";
                             // republish都将刷新使用redis stream自身的id，自定义id无效
                             $header->_id    = '*';
-                            if ($this->requeue($body, $header->toArray())) {
-                                // blocking-retry ack
-                                $this->ack($queueName, $groupName, $this->idsAdd($ids, $id), true);
+                            // 如果错误超过error max count，则存入本地error表
+                            if (
+                                ($errorMaxCount = config('plugin.workbunny.webman-rqueue.app.error_max_count', 0)) > 0 and
+                                $header->_count >= $errorMaxCount
+                            ) {
+                                // 存入
+                                if ($this->tempInsert('error', $queueName, [
+                                    '_header' => $header->toArray(),
+                                    '_body'   => $body
+                                ])) {
+                                    // blocking-retry ack
+                                    $this->ack($queueName, $groupName, $this->idsAdd($ids, $id), true);
+                                }
+                            } else {
+                                if ($this->requeue($body, $header->toArray())) {
+                                    // blocking-retry ack
+                                    $this->ack($queueName, $groupName, $this->idsAdd($ids, $id), true);
 //                                if (!$this->ack($queueName, $groupName, $this->idsAdd($ids, $id))) {
 //                                    $this->idsDel($ids, $id);
 //                                }
+                                }
                             }
                         }
                     }

diff --git a/src/Builders/Traits/MessageTempMethod.php b/src/Builders/Traits/MessageTempMethod.php
@@ -17,7 +17,7 @@ trait MessageTempMethod
     protected ?int $_requeueTimer = null;
 
     protected static array $_tables = [
-        'requeue', 'pending'
+        'requeue', 'pending', 'error'
     ];
 
 

diff --git a/src/config/plugin/workbunny/webman-rqueue/app.php b/src/config/plugin/workbunny/webman-rqueue/app.php
@@ -1,7 +1,10 @@
 <?php
 return [
     'enable' => true,
-
+    // 默认redis连接
     'connection'       => 'default',
-    'requeue_interval' => 0
+    // 本地消息重拾定时器间隔
+    'requeue_interval' => 0,
+    // 最大错误计数
+    'error_max_count'  => 0
 ];