From f7bf93a1f3c05986bc0d2c02c54a997ae91729b8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Wed, 24 Mar 2021 17:20:32 +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/Producer.php | 34 ++++++++++++++++++++++++++++++----
 1 file changed, 30 insertions(+), 4 deletions(-)

diff --git a/Kafka/Producer.php b/Kafka/Producer.php
index 96e5a235..4cc36de0 100644
--- a/Kafka/Producer.php
+++ b/Kafka/Producer.php
@@ -9,8 +9,11 @@ use RdKafka\ProducerTopic;
 use RdKafka\TopicConf;
 use ReflectionException;
 use Snowflake\Abstracts\Component;
+use Snowflake\Abstracts\Config;
+use Snowflake\Core\Json;
 use Snowflake\Event;
 use Snowflake\Exception\ComponentException;
+use Snowflake\Exception\ConfigException;
 use Snowflake\Exception\NotFindClassException;
 use Snowflake\Snowflake;
 
@@ -37,6 +40,10 @@ class Producer extends Component
 	private ?\RdKafka\Producer $producer = null;
 
 
+	/**
+	 * Producer constructor.
+	 * @param array $config
+	 */
 	public function __construct($config = [])
 	{
 		parent::__construct($config);
@@ -81,13 +88,33 @@ class Producer extends Component
 	}
 
 
+	/**
+	 * @param string $topic
+	 * @param array $params
+	 * @param string|null $groupId
+	 * @throws NotFindClassException
+	 * @throws ReflectionException
+	 * @throws ConfigException
+	 */
+	public function dispatch(string $topic, array $params = [], string $groupId = null)
+	{
+		if ($groupId === null || empty($groupId)) {
+			$consumers = Config::get('kafka.consumers.' . $topic);
+			if (empty($consumers)) {
+				$consumers = ['groupId' => $topic . ':' . Snowflake::localhost()];
+			}
+			$groupId = $consumers['groupId'];
+		}
+		$this->setGroupId($groupId)->setTopic($topic)->delivery(Json::encode($params));
+	}
+
+
 	/**
 	 * @param $message
 	 * @param null $key
 	 * @param bool $isAck
 	 * @throws NotFindClassException
 	 * @throws ReflectionException
-	 * @throws ComponentException
 	 * @throws Exception
 	 */
 	public function delivery($message, $key = null, $isAck = false)
@@ -114,18 +141,17 @@ class Producer extends Component
 	}
 
 
-	/** @var ProducerTopic[] $topics  */
+	/** @var ProducerTopic[] $topics */
 	private array $topics = [];
 
 
-
 	/**
 	 * @param $message
 	 * @param $key
 	 */
 	private function push($message, $key)
 	{
-		if(!isset($this->topics[$this->_topic])){
+		if (!isset($this->topics[$this->_topic])) {
 			$this->topics[$this->_topic] = $this->producer->newTopic($this->_topic, $this->topicConf);
 		}
 		$this->topics[$this->_topic]->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);