改名

Kafka/ConsumerInterface.php

@@ -12,13 +12,6 @@ interface ConsumerInterface
 {
 
 
-	/**
-	 * @param Struct $struct
-	 */
-	public function setParams(Struct $struct): void;
-
-
-
 	/**
 	 * @return mixed
 	 */

Kafka/Kafka.php

@@ -124,17 +124,7 @@ class Kafka implements CustomProcess
 
 				$setting = $server->setting['worker_num'];
 
-				$container = Snowflake::app()->get('kafka-container');
-				$handler = $container->getConsumer($topic);
-
-				var_dump($container);
-				if (!empty($handler)) {
-					/** @var ConsumerInterface $data */
-					$data = new $handler();
-					$data->setParams(new Struct($topic, $message));
-
-					$server->sendMessage($data, random_int(0, $setting - 1));
-				}
+				$server->sendMessage(new Message(new Struct($topic, $message)), random_int(0, $setting - 1));
 			} catch (Throwable $exception) {
 				logger()->addError($exception, 'throwable');
 			}

Kafka/Message.php

@@ -0,0 +1,31 @@
+<?php
+
+namespace Kafka;
+
+
+use Server\SInterface\PipeMessage;
+
+/**
+ *
+ */
+class Message implements ConsumerInterface, PipeMessage
+{
+
+
+	/**
+	 * @param Struct $struct
+	 */
+	public function __construct(public Struct $struct)
+	{
+	}
+
+
+	/**
+	 *
+	 */
+	public function process(): void
+	{
+		// TODO: Implement process() method.
+	}
+
+}