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
2020-10-30 11:22:37 +08:00
parent 09d7abdd87
commit f93dc2bd2b
1 changed files with 14 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Producer.php
+14 -15
View File
@@ -31,6 +31,8 @@ class Producer extends Component
private Conf $conf; private Conf $conf;
private TopicConf $topicConf; private TopicConf $topicConf;
private \RdKafka\Producer $producer;
public function __construct($config = []) public function __construct($config = [])
{ {
@@ -91,27 +93,24 @@ class Producer extends Component
$this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason)); $this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
}); });
$event = Snowflake::app()->getEvent();
$event->on(Event::EVENT_AFTER_REQUEST, [$this, 'flush']);
/** @var \RdKafka\Producer $rk */ /** @var \RdKafka\Producer $rk */
$rk = Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]); $this->producer = Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]);
if ($rk->getOutQLen() > 0) { $topic = $this->producer->newTopic($this->_topic, $this->topicConf);
$this->clean($rk, $timeout);
}
$topic = $rk->newTopic($this->_topic, $this->topicConf);
$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key); $topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
$this->clean($rk, $timeout); $this->producer->poll(0);
} }
public function flush()
/**
* @param $rk
* @param $timeout
*/
private function clean($rk, $timeout)
{ {
for ($length = 0; $length < $rk->getOutQLen(); $length++) { while ($this->producer->getOutQLen() > 0) {
$rk->poll(0); $result = $this->producer->flush(100);
if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
break;
}
} }
$rk->flush($timeout);
} }
} }