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-23 11:08:40 +08:00
parent c1c0d862b9
commit 0e0376ce67
1 changed files with 4 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/KafkaClient.php
+4 -3
View File
@@ -29,11 +29,12 @@ class KafkaClient
/** /**
* Producer constructor. * Producer constructor.
* @param string $topic * @param string $topic
* @param string $groupId
* @throws ConfigException
* @throws NotFindClassException * @throws NotFindClassException
* @throws ReflectionException * @throws ReflectionException
* @throws ConfigException
*/ */
public function __construct(public string $topic) public function __construct(public string $topic, public string $groupId)
{ {
$this->conf = di(Conf::class); $this->conf = di(Conf::class);
$this->topicConf = di(TopicConf::class); $this->topicConf = di(TopicConf::class);
@@ -51,7 +52,7 @@ class KafkaClient
throw new ConfigException('Please configure relevant information.'); throw new ConfigException('Please configure relevant information.');
} }
$kafkaConfig->set('metadata.broker.list', $config['brokers']); $kafkaConfig->set('metadata.broker.list', $config['brokers']);
$kafkaConfig->set('group.id', $config['groupId']); $kafkaConfig->set('group.id', $this->groupId);
$kafkaConfig->setErrorCb(function ($kafka, $err, $reason) { $kafkaConfig->setErrorCb(function ($kafka, $err, $reason) {
logger()->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason)); logger()->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
}); });