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-28 11:55:11 +08:00
parent 4a2bcfdde7
commit 3256c331d4
1 changed files with 5 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+5 -5
View File
@@ -81,19 +81,19 @@ class Kafka extends \Snowflake\Process\Process
$objRdKafka = new \RdKafka\Consumer($config); $objRdKafka = new \RdKafka\Consumer($config);
$topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic); $topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic);
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
Timer::tick($conf['interval'] ?? 1000, function () use ($topic) { while (true) {
$this->resolve($topic); $this->resolve($topic, $conf['interval'] ?? 1000);
}); }
} }
/** /**
* @param ConsumerTopic $topic * @param ConsumerTopic $topic
*/ */
private function resolve(ConsumerTopic $topic) private function resolve(ConsumerTopic $topic, $intval)
{ {
try { try {
$message = $topic->consume(0, 10); $message = $topic->consume(0, $intval);
if (empty($message)) { if (empty($message)) {
$this->application->debug('message null.'); $this->application->debug('message null.');
return; return;