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-26 18:01:59 +08:00
parent b65479a677
commit caaf5bba13
1 changed files with 3 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+3 -7
View File
@@ -49,20 +49,16 @@ class Kafka extends \Snowflake\Process\Process
/** /**
* @param Process $process * @param Process $process
* @throws ConfigException * @throws ConfigException
* @throws \RdKafka\Exception
* @throws \Exception * @throws \Exception
*/ */
public function onHandler(Process $process) public function onHandler(Process $process)
{ {
$this->channelListener(); $this->channelListener();
[$config, $conf] = $this->kafkaConfig(); [$config, $conf] = $this->kafkaConfig();
$objRdKafka = new \RdKafka\Consumer($config); $objRdKafka = new KafkaConsumer($config);
$objRdKafka->addBrokers("localhost:9092"); $objRdKafka->subscribe(['test']);
$consumer = $objRdKafka->newTopic('test');
$consumer->consumeStart(0, RD_KAFKA_OFFSET_END);
while (true) { while (true) {
$message = $consumer->consume(0, $conf['metadataRefreshIntervalMs'] ?? 1000); $message = $objRdKafka->consume($conf['metadataRefreshIntervalMs'] ?? 1000);
if (empty($message)) { if (empty($message)) {
continue; continue;
} }