修复队列消费者问题

Author: Yurunsoft

src/Service/BaseQueueConsumer.php

@@ -70,29 +70,33 @@ public function start(?int $co = null)
         $task = function() use($config){
             $queue = $this->imiQueue->getQueue($this->name);
             do {
-                Event::trigger('IMI.QUEUE.CONSUMER.BEFORE_POP', [
-                    'queue' =>  $queue,
-                ], $this, ConsumerBeforePopParam::class);
-                $message = $queue->pop();
-                Event::trigger('IMI.QUEUE.CONSUMER.AFTER_POP', [
-                    'queue'     =>  $queue,
-                    'message'   =>  $message,
-                ], $this, ConsumerAfterPopParam::class);
-                if(null === $message)
-                {
-                    Coroutine::sleep($config->getTimespan());
-                }
-                else
-                {
-                    Event::trigger('IMI.QUEUE.CONSUMER.BEFORE_CONSUME', [
+                try {
+                    Event::trigger('IMI.QUEUE.CONSUMER.BEFORE_POP', [
                         'queue' =>  $queue,
-                        'message'   =>  $message,
-                    ], $this, ConsumerBeforeConsumeParam::class);
-                    $this->consume($message, $queue);
-                    Event::trigger('IMI.QUEUE.CONSUMER.AFTER_CONSUME', [
+                    ], $this, ConsumerBeforePopParam::class);
+                    $message = $queue->pop();
+                    Event::trigger('IMI.QUEUE.CONSUMER.AFTER_POP', [
                         'queue'     =>  $queue,
                         'message'   =>  $message,
-                    ], $this, ConsumerAfterConsumeParam::class);
+                    ], $this, ConsumerAfterPopParam::class);
+                    if(null === $message)
+                    {
+                        Coroutine::sleep($config->getTimespan());
+                    }
+                    else
+                    {
+                        Event::trigger('IMI.QUEUE.CONSUMER.BEFORE_CONSUME', [
+                            'queue' =>  $queue,
+                            'message'   =>  $message,
+                        ], $this, ConsumerBeforeConsumeParam::class);
+                        $this->consume($message, $queue);
+                        Event::trigger('IMI.QUEUE.CONSUMER.AFTER_CONSUME', [
+                            'queue'     =>  $queue,
+                            'message'   =>  $message,
+                        ], $this, ConsumerAfterConsumeParam::class);
+                    }
+                } catch(\Throwable $th) {
+                    App::getBean('ErrorLog')->onException($th);
                 }
             } while($this->working);  
         };
@@ -107,11 +111,7 @@ public function start(?int $co = null)
                  */
                 public function run(ITaskParam $param)
                 {
-                    try {
-                        ($param->getData()['task'])();
-                    } catch(\Throwable $th) {
-                        App::getBean('ErrorLog')->onException($th);
-                    }
+                    ($param->getData()['task'])();
                 }
 
             });