From 1b17d5f78d8afef29b2a5ae042881382979f1e53 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Mr=C2=B7x?= <as2252258@163.com>
Date: Mon, 26 Oct 2020 17:12:16 +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 | 50 +++++++++++++++++++++++++++++++------------------
 1 file changed, 32 insertions(+), 18 deletions(-)

diff --git a/Kafka/Kafka.php b/Kafka/Kafka.php
index cac37839..7f3a1817 100644
--- a/Kafka/Kafka.php
+++ b/Kafka/Kafka.php
@@ -71,7 +71,7 @@ class Kafka extends \Snowflake\Process\Process
 					$this->application->error('No more messages; will wait for more');
 					break;
 				case RD_KAFKA_RESP_ERR__TIMED_OUT:
-					$this->application->error('kafka Timed out');
+					$this->application->error('Kafka Timed out');
 					break;
 				default:
 					throw new \Exception($message->errstr(), $message->err);
@@ -136,31 +136,45 @@ class Kafka extends \Snowflake\Process\Process
 		$conf = new Conf();
 
 		$kafka = SConfig::get('kafka');
-
-		$rdCb = function (KafkaConsumer $kafka, $err, array $partitions = null) {
-			if ($err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) {
-				$kafka->assign($partitions);
-			} else if ($err == RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS) {
-				$kafka->assign(NULL);
-			} else {
-				throw new \Exception($err);
-			}
-		};
-		$conf->setRebalanceCb($rdCb);
+		$conf->setRebalanceCb([$this, 'rebalanced_cb']);
 		$conf->set('group.id', uniqid('kafka'));
+		$conf->set('group.id', 'myConsumerGroup');
+
+		$conf->set('metadata.broker.list', '127.0.0.1');
 		$conf->set('metadata.broker.list', $kafka['brokers']);
+		$conf->set('auto.offset.reset', 'earliest');
+
 		$conf->set('socket.timeout.ms', 300000);
 
 		//多进程和信号
-		if (function_exists('pcntl_sigprocmask')) {
-			pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
-			$conf->set('internal.termination.signal', SIGIO);
-		} else {
-			$conf->set('queue.buffering.max.ms', 1);
-		}
+//		if (function_exists('pcntl_sigprocmask')) {
+//			pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
+//			$conf->set('internal.termination.signal', SIGIO);
+//		} else {
+//			$conf->set('queue.buffering.max.ms', 1);
+//		}
 
 		return [$conf, $kafka];
 	}
 
 
+	/**
+	 * @param KafkaConsumer $kafka
+	 * @param $err
+	 * @param array|null $partitions
+	 * @throws \RdKafka\Exception
+	 * @throws \Exception
+	 */
+	public function rebalanced_cb(KafkaConsumer $kafka, $err, array $partitions = null)
+	{
+		if ($err == RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS) {
+			$kafka->assign($partitions);
+		} else if ($err == RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS) {
+			$kafka->assign(NULL);
+		} else {
+			throw new \Exception($err);
+		}
+	}
+
+
 }