doc:新增异常说明

src/consumer/Consumer.php

@@ -3,7 +3,9 @@
 namespace tm\xls\rocketMq\thinkphp\consumer;
 
 use React\EventLoop\Loop;
+use tm\xls\rocketMq\thinkphp\consumer\contract\ListenerInterface;
 use tm\xls\rocketMq\thinkphp\consumer\message\Message;
+use tm\xls\rocketMq\thinkphp\exception\ConsumerException;
 
 class Consumer
 {
@@ -26,16 +28,20 @@ class Consumer
     public function distribute()
     {
         //事件循环
-        $loop = Loop::get();
+        $loop       = Loop::get();
-        $key  = $this->message->getTopic().$this->message->getTag();
+        $key        = $this->message->getTopic().$this->message->getTag();
         $subscribes = $this->getSubscribe()[$key] ?? [];
-        foreach ($subscribes as $subscribe){
+        foreach ($subscribes as $subscribe) {
-            $loop->addTimer(0, function () use ($subscribe){
+            $loop->addTimer(0, function () use ($subscribe) {
                 $handler = app()->make($subscribe, [
                     $this->message
                 ]);
-                if($handler->enable()){
+                if ($handler instanceof ListenerInterface) {
-                    $handler->handle();
+                    if ($handler->enable()) {
+                        $handler->handle();
+                    }
+                } else {
+                    throw  new ConsumerException('Consumer needs to implement ListenerInterface');
                 }
             });
         }
@@ -47,8 +53,8 @@ class Consumer
      *
      * @return array
      */
-    private function getSubscribe():array
+    private function getSubscribe(): array
     {
-        return  config('rocket_mq.subscribe');
+        return config('rocket_mq.subscribe');
     }
 }

src/exception/ConsumerException.php

@@ -0,0 +1,10 @@
+<?php
+
+namespace tm\xls\rocketMq\thinkphp\exception;
+
+use Exception;
+
+class ConsumerException extends Exception
+{
+
+}

src/producer/Producer.php

@@ -49,7 +49,6 @@ class Producer
                 //todo 失败
             },
         ]);
-
         $promise = $pool->promise();
         $promise->wait();
     }