改名

Kafka/ConsumerInterface.php

@@ -12,11 +12,10 @@ interface ConsumerInterface
 {
 
 
-    /**
-     * @param Struct $struct
-     * @return mixed
-     */
-    public function onHandler(Struct $struct): void;
+	/**
+	 * @return mixed
+	 */
+    public function process(): void;
 
 
 }

Kafka/Kafka.php

@@ -36,7 +36,7 @@ class Kafka implements CustomProcess
      */
     public function __construct(public array $kafkaConfig)
     {
-
+		scan_directory();
     }
 
 
@@ -124,8 +124,6 @@ class Kafka implements CustomProcess
     {
         go(function () use ($topic, $message) {
             try {
-            	var_dump($topic, $message);
-
                 $server = Snowflake::app()->getSwoole();
 
                 $setting = $server->setting['worker_num'];
@@ -183,10 +181,8 @@ class Kafka implements CustomProcess
             return [$conf, $topicConf, $kafka];
         } catch (Throwable $exception) {
             logger()->addError($exception, 'throwable');
-
             return [null, null, null];
         }
-
     }
 
 

Kafka/KafkaProvider.php

@@ -27,7 +27,7 @@ class KafkaProvider extends BaseObject
         if (isset($this->_topics[$topic])) {
             return;
         }
-        $this->_topics[$topic] = $handler;
+        $this->_topics[$topic] = $handler::class;
     }
 
 
@@ -40,18 +40,4 @@ class KafkaProvider extends BaseObject
         return $this->_topics[$topic] ?? null;
     }
 
-
-    /**
-     * @param $topic
-     * @param Struct $struct
-     */
-    public function process($topic, Struct $struct)
-    {
-        $handler = $this->_topics[$topic] ?? null;
-        if (empty($handler)) {
-            return;
-        }
-        call_user_func($handler, $struct);
-    }
-
 }