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:02:38 +08:00
parent 9dd1aa037d
commit 08cc691aff
1 changed files with 9 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+9 -43
View File
@@ -22,50 +22,24 @@ use function Amp\stop;
class Kafka extends \Snowflake\Process\Process class Kafka extends \Snowflake\Process\Process
{ {
/** @var Channel */ protected Channel $channel;
protected $channel;
/**
* @throws ConfigException
*/
public function initConfig()
{
$kafka = SConfig::get('kafka');
$config = ConsumerConfig::getInstance();
$config->setMetadataRefreshIntervalMs(
$kafka['metadataRefreshIntervalMs'] ?? 1000
);
$config->setMetadataBrokerList($kafka['brokers']);
$config->setGroupId($kafka['groupId']);
$config->setBrokerVersion($kafka['version']);
$config->setTopics($kafka['topics']);
$this->channelListener($kafka);
return [new Consumer(), $kafka];
}
/** /**
* @param Process $process * @param Process $process
* @throws ConfigException * @throws ConfigException
* @throws \RdKafka\Exception * @throws \Exception
*/ */
public function onHandler(Process $process) public function onHandler(Process $process)
{ {
// [$consumer, $kafka] = $this->initConfig();
// if ($kafka['debug'] ?? true) {
// $consumer->setLogger(new Logger());
// }
// $consumer->start(function ($topic, $part, $message) {
// $this->channel->push([$topic, $part, $message]);
// });
$this->channelListener(); $this->channelListener();
[$config, $conf] = $this->kafkaConfig(); [$config, $conf] = $this->kafkaConfig();
$consumer = new KafkaConsumer($config); $consumer = new \RdKafka\Consumer($config);
$consumer->subscribe($conf['topics']); $consumer->addBrokers($config['brokers']);
$topic = $consumer->newTopic("test");
while (true) { while (true) {
$message = $consumer->consume($conf['metadataRefreshIntervalMs'] ?? 1000); $message = $topic->consume(0, 1000);
if (empty($message)) { if (empty($message)) {
continue; continue;
} }
@@ -74,10 +48,10 @@ class Kafka extends \Snowflake\Process\Process
$this->channel->push([$message->topic_name, $message->partition, $message]); $this->channel->push([$message->topic_name, $message->partition, $message]);
break; break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF: case RD_KAFKA_RESP_ERR__PARTITION_EOF:
echo "No more messages; will wait for more\n"; $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:
echo "Timed out\n"; $this->application->error('kafka Timed out');
break; break;
default: default:
throw new \Exception($message->errstr(), $message->err); throw new \Exception($message->errstr(), $message->err);
@@ -164,14 +138,6 @@ class Kafka extends \Snowflake\Process\Process
} else { } else {
$conf->set('queue.buffering.max.ms', 1); $conf->set('queue.buffering.max.ms', 1);
} }
//
// $topicConf = new \RdKafka\TopicConf();
// $topicConf->set('auto.commit.enable', 1);
// $topicConf->set('auto.commit.interval.ms', 100);
// $topicConf->set('auto.offset.reset', 'smallest');
// $topicConf->set('offset.store.path', 'kafka_offset.log');
// $conf->set($topicConf);
return [$conf, $kafka]; return [$conf, $kafka];
} }