From 68e0e82c71d300acdf28c8b9372b1aee0bcafc77 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Sun, 14 Feb 2021 21:29: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/Producer.php | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/Kafka/Producer.php b/Kafka/Producer.php
index 3a21cd4f..a11f6ef6 100644
--- a/Kafka/Producer.php
+++ b/Kafka/Producer.php
@@ -5,6 +5,7 @@ namespace Kafka;
 
 use Exception;
 use RdKafka\Conf;
+use RdKafka\ProducerTopic;
 use RdKafka\TopicConf;
 use ReflectionException;
 use Snowflake\Abstracts\Component;
@@ -102,10 +103,21 @@ class Producer extends Component
 	}
 
 
+	/** @var ProducerTopic[] $topics  */
+	private array $topics = [];
+
+
+
+	/**
+	 * @param $message
+	 * @param $key
+	 */
 	private function push($message, $key)
 	{
-		$topic = $this->producer->newTopic($this->_topic, $this->topicConf);
-		$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
+		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);
 		$this->producer->poll(0);
 	}