diff --git a/HttpServer/Events/OnPipeMessage.php b/HttpServer/Events/OnPipeMessage.php
index 61de1d6e..1da51981 100644
--- a/HttpServer/Events/OnPipeMessage.php
+++ b/HttpServer/Events/OnPipeMessage.php
@@ -92,7 +92,9 @@ class OnPipeMessage extends Callback
     {
         [$topic, $rdMessage] = $message['body'];
 
-        call_user_func($message['handler'], new Struct($topic, $rdMessage));
+        $message['handler'][0]->setParams(new Struct($topic, $rdMessage));
+
+        call_user_func($message['handler']);
 
         return 'success';
     }
diff --git a/Kafka/ConsumerInterface.php b/Kafka/ConsumerInterface.php
index 5d7922f4..707db7f9 100644
--- a/Kafka/ConsumerInterface.php
+++ b/Kafka/ConsumerInterface.php
@@ -12,11 +12,14 @@ interface ConsumerInterface
 {
 
 
-	/**
-	 * @param Struct $struct
-	 * @return mixed
-	 */
-	public function onHandler(Struct $struct): void;
+    /**
+     * @param Struct $struct
+     * @return mixed
+     */
+    public function onHandler(Struct $struct): void;
+
+
+    public function setParams(...$params): void;
 
 
 }
diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index 5c83e0ff..e00e7ea7 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -118,6 +118,7 @@ class Kafka extends \Snowflake\Process\Process
                 if (empty($handler)) {
                     return;
                 }
+
                 $message = swoole_serialize(['action' => 'kafka', 'handler' => $handler, 'body' => [$topic, $message]]);
 
                 $server->sendMessage($message, random_int(0, $setting - 1));