药明康德

server/app/Console/Commands/KafkaConsumerCommand.php

@@ -47,6 +47,7 @@ class KafkaConsumerCommand extends Command
         $consumer_type = $this->option('type');
         switch($consumer_type){
             case 'zkh' : app(KafkaService::class)->startZkhConsumer();break;
+            case 'wuxilab' : app(KafkaService::class)->startWuxiLabConsumer();break;
             default : app(KafkaService::class)->startConsumer();break;
         }
     }

server/app/Services/Kafka/KafkaService.php

@@ -99,6 +99,28 @@ class KafkaService
         $consumer->close();
     }
 
+    /**
+     * 药明康德消费
+     */
+    public function startWuxiLabConsumer()
+    {
+        $this->config['topic'] = 'wuxilab';
+        $this->kafkaConsumerConfig = $this->getConsumerConfig($this->config);
+        $consumer = new Consumer($this->kafkaConsumerConfig, function (ConsumeMessage $message) {
+            try{
+                $consumer = $message->getConsumer();
+                $kafkaMessage = json_decode($message->getValue(),true);
+                $consumer->ack($message); // 手动提交
+                app($kafkaMessage['consumer'])->{$kafkaMessage['method']}($kafkaMessage['params']);
+            }catch(\Throwable $exception){
+                $this->errLog($exception,$this->kafkaConsumerConfig->getGroupInstanceId().'_consumer');
+            }
+            $consumer->stop();
+        });
+        $consumer->start();
+        $consumer->close();
+    }
+
     protected function errLog($exception, $errorType)
     {
         SimpleLogs::writeLog([