save

workbunny · Feb 12, 2024 · bf0763e · bf0763e
1 parent 549ad3c
commit bf0763e
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 9 deletions.
diff --git a/composer.json b/composer.json
@@ -22,13 +22,13 @@
     "webman/console": "^1.0",
     "illuminate/redis": "^9.0 | ^10.0",
     "illuminate/events": "^9.0 | ^10.0",
-    "illuminate/database": "^9.0 | ^10.0"
+    "illuminate/database": "^9.0 | ^10.0",
+    "monolog/monolog": "^2.0 | ^3.0"
   },
   "require-dev": {
     "workerman/webman-framework": "^1.0",
     "symfony/var-dumper": "^6.0",
-    "phpunit/phpunit": "^9.6",
-    "monolog/monolog": "^2.0"
+    "phpunit/phpunit": "^9.6"
   },
   "autoload": {
     "psr-4": {

diff --git a/src/Builders/AdaptiveBuilder.php b/src/Builders/AdaptiveBuilder.php
@@ -7,7 +7,6 @@
 use RedisException;
 use support\Log;
 use Workbunny\WebmanRqueue\Builders\Traits\AdaptiveTimerMethod;
-use Workbunny\WebmanRqueue\Builders\Traits\MessageQueueMethod;
 use Workbunny\WebmanRqueue\Exceptions\WebmanRqueueException;
 use Workerman\Timer;
 use Workerman\Worker;
@@ -64,7 +63,7 @@ public function onWorkerStart(Worker $worker): void
             $this->adaptiveTimerCreate($this->timerInterval / 1000, function () use($worker) {
                 try {
                     // consume
-                    $this->consume($worker);
+                   return $this->consume($worker);
                 } catch (WebmanRqueueException $exception) {
                     Log::channel('plugin.workbunny.webman-rqueue.warning')?->warning('Consume exception. ', [
                         'message' => $exception->getMessage(), 'code' => $exception->getCode(),
@@ -76,6 +75,7 @@ public function onWorkerStart(Worker $worker): void
                         'message' => $exception->getMessage(), 'code' => $exception->getCode()
                     ]);
                 }
+                return false;
             });
         }
     }
@@ -86,16 +86,16 @@ public function onWorkerStop(Worker $worker): void
         if($this->getConnection()) {
             try {
                 $this->getConnection()->client()->close();
-            }catch (RedisException $e) {
+            } catch (RedisException $e) {
                 echo $e->getMessage() . PHP_EOL;
             }
         }
-        // 移除自适应
-        $this->adaptiveTimerDelete();
-        //
+        // 移除pending检查定时器
         if($this->getPendingTimer()) {
             Timer::del($this->getPendingTimer());
         }
+        // 移除自适应
+        $this->adaptiveTimerDelete();
     }
 
     /** @inheritDoc */

diff --git a/src/Builders/Traits/AdaptiveTimerMethod.php b/src/Builders/Traits/AdaptiveTimerMethod.php
@@ -139,6 +139,7 @@ public function adaptiveTimerCreate(int $millisecond, Closure $func, mixed ...$a
             EventInterface::EV_TIMER,
             $callback = function (...$args) use ($func, $millisecond, $id, &$callback)
             {
+                // 获取毫秒时间戳
                 $nowMilliTimestamp = intval(microtime(true) * 1000);
                 if (\call_user_func($func, ...$args)) {
                     self::$lastMessageMilliTimestamp = $nowMilliTimestamp;