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-12 12:56:07 +08:00
parent 0a9be5595a
commit 14a026546a
1 changed files with 6 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+6 -5
View File
@@ -38,7 +38,7 @@ class Kafka extends \Snowflake\Process\Process
$config->setGroupId($kafka['groupId']); $config->setGroupId($kafka['groupId']);
$config->setBrokerVersion($kafka['version']); $config->setBrokerVersion($kafka['version']);
$config->setTopics($kafka['topics']); $config->setTopics($kafka['topics']);
$this->channelListener(); $this->channelListener($kafka);
return [new Consumer(), $kafka]; return [new Consumer(), $kafka];
} }
@@ -62,13 +62,14 @@ class Kafka extends \Snowflake\Process\Process
/** /**
* 监听通道数据传递 * 监听通道数据传递
* @param $config
*/ */
public function channelListener() public function channelListener($config)
{ {
$this->channel = new Channel(1000); $this->channel = new Channel($config['size'] ?? 100);
Coroutine::create(function () { Coroutine::create(function () use ($config) {
$group = new WaitGroup(); $group = new WaitGroup();
for ($i = 0; $i < 1000; $i++) { for ($i = 0; $i < $config['size'] ?? 100; $i++) {
$group->add(); $group->add();
go(function () use ($group) { go(function () use ($group) {
while ([$topic, $part, $message] = $this->channel->pop()) { while ([$topic, $part, $message] = $this->channel->pop()) {