From e03f8427ed26b73569660f91f792a86133684bf2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Thu, 5 Aug 2021 16:04:59 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Kafka/Kafka.php | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index 5917bc36..87bd0199 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -11,6 +11,7 @@ use RdKafka\Exception; use RdKafka\KafkaConsumer; use RdKafka\TopicConf; use ReflectionException; +use Server\ServerManager; use Server\SInterface\CustomProcess; use Snowflake\Abstracts\Config; use Snowflake\Exception\ConfigException; @@ -133,13 +134,11 @@ class Kafka implements CustomProcess $container = Snowflake::app()->get('kafka-container'); $handler = $container->getConsumer($topic); - if (empty($handler)) { - return; + if (!empty($handler)) { + $data = new Struct($topic, $message); + + $server->sendMessage(new $handler($data), random_int(0, $setting - 1)); } - - $message = swoole_serialize(['action' => 'kafka', 'handler' => $handler, 'body' => [$topic, $message]]); - - $server->sendMessage($message, random_int(0, $setting - 1)); } catch (Throwable $exception) { logger()->addError($exception, 'throwable'); }