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-08-03 18:18:09 +08:00
parent 324db2fa3f
commit 5b28c52a10
43 changed files with 755 additions and 653 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Producer.php
+111 -136
View File
@@ -7,12 +7,9 @@ use Exception;
use RdKafka\Conf;
use RdKafka\ProducerTopic;
use RdKafka\TopicConf;
use ReflectionException;
use Snowflake\Abstracts\Component;
use Snowflake\Abstracts\Config;
use Snowflake\Event;
use Snowflake\Exception\ConfigException;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;
/**
@@ -30,77 +27,74 @@ class Producer extends Component
{
private Conf $conf;
private TopicConf $topicConf;
private Conf $conf;
private TopicConf $topicConf;
private ?\RdKafka\Producer $producer = null;
private ?\RdKafka\Producer $producer = null;
private bool $isAck = true;
private bool $isAck = true;
/**
* Producer constructor.
* @param array $config
* @throws Exception
*/
public function __construct(array $config = [])
{
parent::__construct($config);
if (!class_exists(Conf::class)) {
return;
}
$this->conf = new Conf();
$this->topicConf = new TopicConf();
$this->conf->setErrorCb(function ($kafka, $err, $reason) {
$this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
});
}
/**
* Producer constructor.
* @param array $config
* @throws Exception
*/
public function __construct(array $config = [])
{
parent::__construct($config);
if (!class_exists(Conf::class)) {
return;
}
$this->conf = new Conf();
$this->topicConf = new TopicConf();
$this->conf->setErrorCb(function ($kafka, $err, $reason) {
$this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
});
}
/**
* @param string $servers
* @return Producer
*/
public function setBrokers(string $servers): static
{
$this->conf->set('metadata.broker.list', $servers);
return $this;
}
public function setBrokers(string $servers): static
{
$this->conf->set('metadata.broker.list', $servers);
return $this;
}
/**
* @param string $groupId
* @return Producer
*/
public function setGroupId(string $groupId): static
{
$this->conf->set('group.id', $groupId);
return $this;
}
/**
* @param string $groupId
* @return Producer
*/
public function setGroupId(string $groupId): static
{
$this->conf->set('group.id', $groupId);
return $this;
}
/**
* @param string $topic
* @param array $params
* @param string|null $groupId
* @throws Exception
*/
public function dispatch(string $topic, array $params = [], string $groupId = null)
{
$this->beforePushMessage($topic, $groupId);
$this->sendMessage($topic, [$params]);
}
/**
* @param string $topic
* @param array $params
* @param string|null $groupId
* @throws Exception
*/
public function dispatch(string $topic, array $params = [], string $groupId = null)
{
$this->beforePushMessage($topic, $groupId);
$this->sendMessage($topic, [$params]);
}
/**
* @return \RdKafka\Producer
* @throws Exception
*/
private function getProducer(): \RdKafka\Producer
{
$pool = Snowflake::app()->getChannel();
return $pool->pop(\RdKafka\Producer::class, function () {
return Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]);
});
}
/**
* @return \RdKafka\Producer
* @throws Exception
*/
private function getProducer(): \RdKafka\Producer
{
return Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]);
}
/**
@@ -109,13 +103,10 @@ class Producer extends Component
* @return ProducerTopic
* @throws Exception
*/
private function getProducerTopic(\RdKafka\Producer $producer, $topic): ProducerTopic
{
$pool = Snowflake::app()->getChannel();
return $pool->pop($topic . '::' . ProducerTopic::class, function () use ($producer, $topic) {
return $producer->newTopic($topic, $this->topicConf);
});
}
private function getProducerTopic(\RdKafka\Producer $producer, $topic): ProducerTopic
{
return $producer->newTopic($topic, $this->topicConf);
}
/**
@@ -126,12 +117,12 @@ class Producer extends Component
* @throws ConfigException
* @throws Exception
*/
public function batch(string $toPic, ?string $key, array $data, ?string $groupId = null)
{
$this->beforePushMessage($toPic, $groupId);
public function batch(string $toPic, ?string $key, array $data, ?string $groupId = null)
{
$this->beforePushMessage($toPic, $groupId);
$this->sendMessage($toPic, $data, $key);
}
$this->sendMessage($toPic, $data, $key);
}
/**
@@ -140,23 +131,23 @@ class Producer extends Component
* @throws ConfigException
* @throws Exception
*/
private function beforePushMessage($topic, $groupId): void
{
$consumers = Config::get('kafka.producers.' . $topic);
if (empty($consumers) || !is_array($consumers)) {
throw new Exception('You need set kafka.producers config');
}
if (!isset($consumers['brokers'])) {
throw new Exception('You need set brokers config.');
}
if (!empty($groupId)) {
$consumers['groupId'] = $groupId;
} else if (!isset($consumers['groupId'])) {
$consumers['groupId'] = $topic . ':' . Snowflake::localhost();
}
$this->setGroupId($consumers['groupId']);
$this->setBrokers($consumers['brokers']);
}
private function beforePushMessage($topic, $groupId): void
{
$consumers = Config::get('kafka.producers.' . $topic);
if (empty($consumers) || !is_array($consumers)) {
throw new Exception('You need set kafka.producers config');
}
if (!isset($consumers['brokers'])) {
throw new Exception('You need set brokers config.');
}
if (!empty($groupId)) {
$consumers['groupId'] = $groupId;
} else if (!isset($consumers['groupId'])) {
$consumers['groupId'] = $topic . ':' . Snowflake::localhost();
}
$this->setGroupId($consumers['groupId']);
$this->setBrokers($consumers['brokers']);
}
/**
@@ -165,58 +156,42 @@ class Producer extends Component
* @param string $key
* @throws Exception
*/
private function sendMessage(string $topic, array $message, string $key = '')
{
$producer = $this->getProducer();
$producerTopic = $this->getProducerTopic($producer, $topic);
if ($this->isAck) {
$this->flush($producer);
}
foreach ($message as $value) {
$producerTopic->produce(RD_KAFKA_PARTITION_UA, 0, swoole_serialize($value), $key);
$producer->poll(0);
}
$this->flush($producer);
$this->recover($topic, $producer, $producerTopic);
}
/**
* @param bool $ack
*/
public function setAsk(bool $ack){
$this->isAck = $ack;
$this->topicConf->set('request.required.acks', $this->isAck ? '1' : '0');
}
private function sendMessage(string $topic, array $message, string $key = '')
{
$producer = $this->getProducer();
$producerTopic = $this->getProducerTopic($producer, $topic);
if ($this->isAck) {
$this->flush($producer);
}
foreach ($message as $value) {
$producerTopic->produce(RD_KAFKA_PARTITION_UA, 0, swoole_serialize($value), $key);
$producer->poll(0);
}
$this->flush($producer);
}
/**
* @param string $topic
* @param \RdKafka\Producer $producer
* @param ProducerTopic $producerTopic
* @throws Exception
* @param bool $ack
*/
private function recover(string $topic, \RdKafka\Producer $producer, ProducerTopic $producerTopic)
{
$channel = Snowflake::app()->getChannel();
$channel->push($producerTopic, $topic . '::' . ProducerTopic::class);
$channel->push($producer, \RdKafka\Producer::class);
}
public function setAsk(bool $ack)
{
$this->isAck = $ack;
$this->topicConf->set('request.required.acks', $this->isAck ? '1' : '0');
}
/**
* @param \RdKafka\Producer $producer
*/
public function flush(\RdKafka\Producer $producer)
{
while ($producer->getOutQLen() > 0) {
$result = $producer->flush(100);
if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
break;
}
}
}
/**
* @param \RdKafka\Producer $producer
*/
public function flush(\RdKafka\Producer $producer)
{
while ($producer->getOutQLen() > 0) {
$result = $producer->flush(100);
if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
break;
}
}
}
}