as2252258/kiri-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

改名

Browse Source
This commit is contained in:
Mr·x
2021-02-14 21:29:59 +08:00
parent 294258994b
commit 68e0e82c71
1 changed files with 14 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Producer.php
+14 -2
View File
@@ -5,6 +5,7 @@ namespace Kafka;
use Exception; use Exception;
use RdKafka\Conf; use RdKafka\Conf;
use RdKafka\ProducerTopic;
use RdKafka\TopicConf; use RdKafka\TopicConf;
use ReflectionException; use ReflectionException;
use Snowflake\Abstracts\Component; 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) private function push($message, $key)
{ {
$topic = $this->producer->newTopic($this->_topic, $this->topicConf); if(!isset($this->topics[$this->_topic])){
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key); $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); $this->producer->poll(0);
} }