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 17:12:16 +08:00
parent ad0d48f775
commit 1b17d5f78d
1 changed files with 32 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+32 -18
View File
@@ -71,7 +71,7 @@ class Kafka extends \Snowflake\Process\Process
$this->application->error('No more messages; will wait for more'); $this->application->error('No more messages; will wait for more');
break; break;
case RD_KAFKA_RESP_ERR__TIMED_OUT: case RD_KAFKA_RESP_ERR__TIMED_OUT:
$this->application->error('kafka Timed out'); $this->application->error('Kafka Timed out');
break; break;
default: default:
throw new \Exception($message->errstr(), $message->err); throw new \Exception($message->errstr(), $message->err);
@@ -136,31 +136,45 @@ class Kafka extends \Snowflake\Process\Process
$conf = new Conf(); $conf = new Conf();
$kafka = SConfig::get('kafka'); $kafka = SConfig::get('kafka');
$conf->setRebalanceCb([$this, 'rebalanced_cb']);
$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->set('group.id', uniqid('kafka')); $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('metadata.broker.list', $kafka['brokers']);
$conf->set('auto.offset.reset', 'earliest');
$conf->set('socket.timeout.ms', 300000); $conf->set('socket.timeout.ms', 300000);
//多进程和信号 //多进程和信号
if (function_exists('pcntl_sigprocmask')) { // if (function_exists('pcntl_sigprocmask')) {
pcntl_sigprocmask(SIG_BLOCK, array(SIGIO)); // pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
$conf->set('internal.termination.signal', SIGIO); // $conf->set('internal.termination.signal', SIGIO);
} else { // } else {
$conf->set('queue.buffering.max.ms', 1); // $conf->set('queue.buffering.max.ms', 1);
} // }
return [$conf, $kafka]; 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);
}
}
} }