openapi

server/app/Services/Kafka/KafkaService.php

@@ -64,12 +64,14 @@ class KafkaService
             $consumer = $message->getConsumer();
             $kafkaMessage = json_decode($message->getValue(),true);
 
+            if(!empty($kafkaMessage)){
                 try{
                     app($kafkaMessage['consumer'])->{$kafkaMessage['method']}($kafkaMessage['params']);
                     $consumer->ack($message); // 手动提交
                 }catch(\Exception $exception){
                     $this->errLog($exception);
                 }
+            }
 
             $consumer->stop();
         });