From a0688a587491b1d9c90c345708584789ae002105 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Tue, 12 Jan 2021 17:29:48 +0800
Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 Kafka/Kafka.php | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index 5b85f395..3f35e1f8 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -52,7 +52,7 @@ class Kafka extends \Snowflake\Process\Process
 			$objRdKafka = new Consumer($config);
 			$topic = $objRdKafka->newTopic($kafkaServer['topic'], $topic);
 
-			$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
+			$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);
 			while (true) {
 				$this->resolve($topic, $conf['interval'] ?? 1000);
 			}
@@ -151,19 +151,18 @@ class Kafka extends \Snowflake\Process\Process
 		if (function_exists('pcntl_sigprocmask')) {
 			pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
 			$conf->set('internal.termination.signal', (string)SIGIO);
-//		} else {
-//			$conf->set('queue.buffering.max.ms', '1');
 		}
 
 		$topicConf = new TopicConf();
 		try {
 			$topicConf->set('auto.commit.enable', '1');
 			$topicConf->set('auto.commit.interval.ms', '100');
-			//smallest：简单理解为从头开始消费，largest：简单理解为从最新的开始消费
+
+			//smallest：简单理解为从头开始消费，
+			//largest：简单理解为从最新的开始消费
 			$topicConf->set('auto.offset.reset', 'smallest');
 			$topicConf->set('offset.store.path', 'kafka_offset.log');
 			$topicConf->set('offset.store.method', 'broker');
-//			$topicConf->set('allow.auto.create.topics', 'true');
 		} catch (Throwable $exception) {
 			var_dump($exception->getMessage());
 		}