modify

Kafka/Kafka.php

@@ -104,7 +104,14 @@ class Kafka extends \Snowflake\Process\Process
 
                 $setting = $server->setting['worker_num'];
 
-                $message = swoole_serialize(['action' => 'kafka', 'body' => [$topic, $message]]);
+                /** @var TaskContainer $container */
+                $container = Snowflake::app()->get('kafka-container');
+                $handler = $container->getConsumer($topic);
+
+                if (empty($handler)) {
+                    return;
+                }
+                $message = swoole_serialize(['action' => 'kafka', 'handler' => $handler, 'body' => [$topic, $message]]);
 
                 $server->sendMessage($message, random_int(0, $setting - 1));
             } catch (Throwable $exception) {

Kafka/TaskContainer.php

@@ -31,6 +31,16 @@ class TaskContainer extends BaseObject
     }
 
 
+    /**
+     * @param $topic
+     * @return mixed|null
+     */
+    public function getConsumer(string $topic)
+    {
+        return $this->_topics[$topic] ?? null;
+    }
+
+
     /**
      * @param $topic
      * @param \Kafka\Struct $struct