From 6fbb0dea760e5905cf251242e951a743e72304db Mon Sep 17 00:00:00 2001 From: "as2252258@163.com" Date: Sun, 11 Apr 2021 17:18:54 +0800 Subject: [PATCH] modify --- Annotation/Kafka.php | 12 +- HttpServer/Events/OnPipeMessage.php | 46 ++++- Kafka/Kafka.php | 259 ++++++++++++++-------------- Kafka/TaskContainer.php | 62 +++++++ 4 files changed, 245 insertions(+), 134 deletions(-) create mode 100644 Kafka/TaskContainer.php diff --git a/Annotation/Kafka.php b/Annotation/Kafka.php index 47344b8e..82698cf8 100644 --- a/Annotation/Kafka.php +++ b/Annotation/Kafka.php @@ -4,6 +4,9 @@ namespace Annotation; +use Kafka\ConsumerInterface; +use Kafka\TaskContainer; + /** * Class Kafka * @package Annotation @@ -16,7 +19,7 @@ namespace Annotation; * Kafka constructor. * @param string $topic */ - public function __construct(public string $topic, public string $groupId, public ?string $brokers = null) + public function __construct(public string $topic) { } @@ -28,6 +31,13 @@ namespace Annotation; */ public function execute(array $handler): mixed { + if (!($handler[0] instanceof ConsumerInterface)) { + return false; + } + + $container = TaskContainer::getInstance(); + $container->addConsumer($this->topic, [$handler[0], 'onHandler']); + return parent::execute($handler); // TODO: Change the autogenerated stub } diff --git a/HttpServer/Events/OnPipeMessage.php b/HttpServer/Events/OnPipeMessage.php index 6627c42e..ba385cbf 100644 --- a/HttpServer/Events/OnPipeMessage.php +++ b/HttpServer/Events/OnPipeMessage.php @@ -7,6 +7,9 @@ namespace HttpServer\Events; use Annotation\Loader; use Exception; use HttpServer\Abstracts\Callback; +use Kafka\ConsumerInterface; +use Kafka\Struct; +use Kafka\TaskContainer; use Snowflake\Event; use Snowflake\Exception\ComponentException; use Snowflake\Snowflake; @@ -28,8 +31,47 @@ class OnPipeMessage extends Callback */ public function onHandler(Server $server, int $src_worker_id, $message) { - $events = Snowflake::app()->getEvent(); - $events->trigger(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]); + try { + $swoole_unserialize = swoole_unserialize($message); + match ($swoole_unserialize['action'] ?? null) { + 'kafka' => $this->onKafkaWorker($swoole_unserialize), + default => $this->onMessageWorker($server, $src_worker_id, $message) + }; + } catch (\Throwable $exception) { + $this->addError($exception); + } finally { + fire(Event::SYSTEM_RESOURCE_RELEASES); + } } + + /** + * @param $server + * @param $src_worker_id + * @param $message + * @throws \Exception + */ + private function onMessageWorker($server, $src_worker_id, $message) + { + fire(Event::PIPE_MESSAGE, [$server, $src_worker_id, $message]); + + return 'success'; + } + + + /** + * @param array $message + * @throws \ReflectionException + * @throws \Snowflake\Exception\NotFindClassException + */ + private function onKafkaWorker(array $message) + { + [$topic, $message] = $message['body']; + + $container = TaskContainer::getInstance(); + $container->process($topic, new Struct($topic, $message)); + return 'success'; + } + + } diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index 2d5ea01c..ce7a10db 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -10,6 +10,7 @@ use RdKafka\ConsumerTopic; use RdKafka\Exception; use RdKafka\KafkaConsumer; use RdKafka\TopicConf; +use Snowflake\Core\Json; use Snowflake\Snowflake; use Swoole\Coroutine\Channel; use Swoole\Process; @@ -22,160 +23,156 @@ use Throwable; class Kafka extends \Snowflake\Process\Process { - protected Channel $channel; + protected Channel $channel; - private int $maxLength = 5000; + private int $maxLength = 5000; - /** - * @param Process $process - * @throws \Exception - */ - public function onHandler(Process $process): void - { - $this->waite(swoole_unserialize($process->read())); - } + /** + * @param Process $process + * @throws \Exception + */ + public function onHandler(Process $process): void + { + $this->waite(swoole_unserialize($process->read())); + } - /** - * @param array $kafkaServer - * @throws \Exception - */ - private function waite(array $kafkaServer) - { - try { - name($this->pid, 'Kafka Consumer ' . $kafkaServer['topic']); + /** + * @param array $kafkaServer + * @throws \Exception + */ + private function waite(array $kafkaServer) + { + try { + name($this->pid, 'Kafka Consumer ' . $kafkaServer['topic']); - [$config, $topic, $conf] = $this->kafkaConfig($kafkaServer); - if (empty($config) && empty($topic) && empty($conf)) { - return; - } - $objRdKafka = new Consumer($config); - $topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic); + [$config, $topic, $conf] = $this->kafkaConfig($kafkaServer); + if (empty($config) && empty($topic) && empty($conf)) { + return; + } + $objRdKafka = new Consumer($config); + $topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic); - $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); - do { - $this->resolve($topic, $conf['interval'] ?? 1000); - } while (true); - } catch (Throwable $exception) { - logger()->addError($exception, 'throwable'); - } - } + $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); + do { + $this->resolve($topic, $conf['interval'] ?? 1000); + } while (true); + } catch (Throwable $exception) { + logger()->addError($exception, 'throwable'); + } + } - /** - * @param ConsumerTopic $topic - * @param $interval - * @throws \Exception - */ - private function resolve(ConsumerTopic $topic, $interval) - { - try { - $message = $topic->consume(0, $interval); - if (empty($message)) { - return; - } - if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) { - $this->handlerExecute($message->topic_name, $message); - } else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { - $this->application->warning('No more messages; will wait for more'); - } else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) { - $this->application->error('Kafka Timed out'); - } else { - $this->application->error($message->errstr()); - } - } catch (Throwable $exception) { - logger()->addError($exception, 'throwable'); - } - } + /** + * @param ConsumerTopic $topic + * @param $interval + * @throws \Exception + */ + private function resolve(ConsumerTopic $topic, $interval) + { + try { + $message = $topic->consume(0, $interval); + if (empty($message)) { + return; + } + if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) { + $this->handlerExecute($message->topic_name, $message); + } else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { + $this->application->warning('No more messages; will wait for more'); + } else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) { + $this->application->error('Kafka Timed out'); + } else { + $this->application->error($message->errstr()); + } + } catch (Throwable $exception) { + logger()->addError($exception, 'throwable'); + } + } - /** - * @param $topic - * @param $message - * @throws \Exception - */ - protected function handlerExecute($topic, $message) - { - go(function () use ($topic, $message) { - try { - $topic = str_replace('-', '_', $topic); + /** + * @param $topic + * @param $message + * @throws \Exception + */ + protected function handlerExecute($topic, $message) + { + go(function () use ($topic, $message) { + try { + $server = Snowflake::app()->getSwoole(); - $namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer'; - if (!class_exists($namespace)) { - return; - } - $class = Snowflake::createObject($namespace); - if (!($class instanceof ConsumerInterface)) { - return; - } - $class->onHandler(Snowflake::createObject(Struct::class, [$topic, $message])); - } catch (Throwable $exception) { - logger()->addError($exception, 'throwable'); - } - }); - } + $setting = $server->setting['worker_num']; + + $message = swoole_serialize(['action' => 'kafka', 'body' => [$topic, $message]]); + + $server->sendMessage($message, random_int(0, $setting)); + } catch (Throwable $exception) { + logger()->addError($exception, 'throwable'); + } + }); + } - /** - * @param $kafka - * @return array - * @throws \Exception - */ - private function kafkaConfig($kafka): array - { - try { - $conf = new Conf(); - $conf->setRebalanceCb([$this, 'rebalanced_cb']); - $conf->set('group.id', $kafka['groupId']); - $conf->set('metadata.broker.list', $kafka['brokers']); - $conf->set('socket.timeout.ms', '30000'); + /** + * @param $kafka + * @return array + * @throws \Exception + */ + private function kafkaConfig($kafka): array + { + try { + $conf = new Conf(); + $conf->setRebalanceCb([$this, 'rebalanced_cb']); + $conf->set('group.id', $kafka['groupId']); + $conf->set('metadata.broker.list', $kafka['brokers']); + $conf->set('socket.timeout.ms', '30000'); - $this->application->debug('kafka listen groupId ' . $kafka['groupId']); - $this->application->debug('kafka listen brokers ' . $kafka['brokers']); + $this->application->debug('kafka listen groupId ' . $kafka['groupId']); + $this->application->debug('kafka listen brokers ' . $kafka['brokers']); - if (function_exists('pcntl_sigprocmask')) { - pcntl_sigprocmask(SIG_BLOCK, array(SIGIO)); - $conf->set('internal.termination.signal', (string)SIGIO); - } + if (function_exists('pcntl_sigprocmask')) { + pcntl_sigprocmask(SIG_BLOCK, array(SIGIO)); + $conf->set('internal.termination.signal', (string)SIGIO); + } - $topicConf = new TopicConf(); - $topicConf->set('auto.commit.enable', '1'); - $topicConf->set('auto.commit.interval.ms', '100'); + $topicConf = new TopicConf(); + $topicConf->set('auto.commit.enable', '1'); + $topicConf->set('auto.commit.interval.ms', '100'); - //smallest：简单理解为从头开始消费， - //largest：简单理解为从最新的开始消费 - $topicConf->set('auto.offset.reset', 'smallest'); - $topicConf->set('offset.store.path', 'kafka_offset.log'); - $topicConf->set('offset.store.method', 'broker'); + //smallest：简单理解为从头开始消费， + //largest：简单理解为从最新的开始消费 + $topicConf->set('auto.offset.reset', 'smallest'); + $topicConf->set('offset.store.path', 'kafka_offset.log'); + $topicConf->set('offset.store.method', 'broker'); - return [$conf, $topicConf, $kafka]; - } catch (Throwable $exception) { - logger()->addError($exception, 'throwable'); + return [$conf, $topicConf, $kafka]; + } catch (Throwable $exception) { + logger()->addError($exception, 'throwable'); - return [null, null, null]; - } + return [null, null, null]; + } - } + } - /** - * @param KafkaConsumer $kafka - * @param $err - * @param array|null $partitions - * @throws Exception - * @throws \Exception - */ - public function rebalanced_cb(KafkaConsumer $kafka, $err, array $partitions = null) - { - if ($err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) { - $kafka->assign($partitions); - } else if ($err == RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS) { - $kafka->assign(NULL); - } else { - throw new \Exception($err); - } - } + /** + * @param KafkaConsumer $kafka + * @param $err + * @param array|null $partitions + * @throws Exception + * @throws \Exception + */ + public function rebalanced_cb(KafkaConsumer $kafka, $err, array $partitions = null) + { + if ($err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) { + $kafka->assign($partitions); + } else if ($err == RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS) { + $kafka->assign(NULL); + } else { + throw new \Exception($err); + } + } } diff --git a/Kafka/TaskContainer.php b/Kafka/TaskContainer.php new file mode 100644 index 00000000..b7b660c4 --- /dev/null +++ b/Kafka/TaskContainer.php @@ -0,0 +1,62 @@ +_topics[$topic])) { + return; + } + $this->_topics[$topic] = $handler; + } + + + /** + * @param $topic + * @param \Kafka\Struct $struct + */ + public function process($topic, Struct $struct) + { + $handler = $this->_topics[$topic] ?? null; + if (empty($handler)) { + return; + } + call_user_func($handler, $struct); + } + +}