modify

2021-04-11 17:18:54 +08:00
parent 364c638d81
commit 6fbb0dea76
4 changed files with 245 additions and 134 deletions
@@ -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
    }

@@ -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';
+    }
+
+
 }
@@ -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);
+        }
+    }


 }
@@ -0,0 +1,62 @@
+<?php
+
+
+namespace Kafka;
+
+
+use Snowflake\Abstracts\BaseObject;
+
+
+/**
+ * Class TaskContainer
+ * @package Kafka
+ */
+class TaskContainer extends BaseObject
+{
+
+
+    private array $_topics = [];
+
+
+    private static TaskContainer $container;
+
+
+    /**
+     * @return \Kafka\TaskContainer
+     */
+    public static function getInstance(): TaskContainer
+    {
+        if (!(static::$container instanceof TaskContainer)) {
+            static::$container = new TaskContainer();
+        }
+        return static::$container;
+    }
+
+
+    /**
+     * @param $topic
+     * @param $handler
+     */
+    public function addConsumer($topic, $handler)
+    {
+        if (isset($this->_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);
+    }
+
+}