diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php index 61c788ef..7ab052a5 100644 --- a/Kafka/Kafka.php +++ b/Kafka/Kafka.php @@ -82,8 +82,8 @@ class Kafka extends \Snowflake\Process\Process $objRdKafka = new \RdKafka\Consumer($config); $topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic); $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED); + var_dump($topic); Timer::tick($conf['interval'] ?? 1000, function () use ($topic) { - var_dump($topic); $this->resolve($topic); }); } @@ -95,7 +95,7 @@ class Kafka extends \Snowflake\Process\Process private function resolve(ConsumerTopic $topic) { try { - $message = $topic->consume(0, 0); + $message = $topic->consume(0, 10); if (empty($message)) { $this->application->debug('message null.'); return;