kafka

Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，有如下特性： 通過(guò)O(1)的磁盤數(shù)據(jù)結(jié)構(gòu)提供消息的持久化，這種結(jié)構(gòu)對(duì)于即使數(shù)以TB的消息存儲(chǔ)也能夠保持長(zhǎng)時(shí)間的穩(wěn)定性能。 高吞吐量：即使是非常普通的硬件Kafka也可以支持每秒數(shù)百萬(wàn)的消息。 支持通過(guò)Kafka服務(wù)器和消費(fèi)機(jī)集群來(lái)分區(qū)消息。 支持Hadoop并行數(shù)據(jù)加載。

本項(xiàng)目代碼參考自 https://github.com/weiboad/kafka-php

組件要求

• php: >=7.1.0
• ext-swoole: ^4.4.5
• easyswoole/component: ^2.0
• easyswoole/spl: ^1.1

安裝方法

composer require easyswoole/kafka

倉(cāng)庫(kù)地址

easyswoole/kafka

基本使用

注冊(cè)kafka服務(wù)

namespace EasySwoole\EasySwoole;

use App\Producer\Process as ProducerProcess;
use App\Consumer\Process as ConsumerProcess;
use EasySwoole\EasySwoole\Swoole\EventRegister;
use EasySwoole\EasySwoole\AbstractInterface\Event;
use EasySwoole\Http\Request;
use EasySwoole\Http\Response;

class EasySwooleEvent implements Event
{

    public static function initialize()
    {
        // TODO: Implement initialize() method.
        date_default_timezone_set('Asia/Shanghai');
    }

    public static function mainServerCreate(EventRegister $register)
    {
        // TODO: Implement mainServerCreate() method.
        // 生產(chǎn)者
        \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new ProducerProcess());
        // 消費(fèi)者
        \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new ConsumerProcess());
    }
}

生產(chǎn)者

namespace App\Producer;

use EasySwoole\Component\Process\AbstractProcess;
use EasySwoole\Kafka\Config\ProducerConfig;
use EasySwoole\Kafka\Kafka;

class Process extends AbstractProcess
{
    protected function run($arg)
    {
        go(function () {
            $config = new ProducerConfig();
            $config->setMetadataBrokerList('127.0.0.1:9092,127.0.0.1:9093');
            $config->setBrokerVersion('0.9.0');
            $config->setRequiredAck(1);

            $kafka = new Kafka($config);
            $result = $kafka->producer()->send([
                [
                    'topic' => 'test',
                    'value' => 'message--',
                    'key'   => 'key--',
                ],
            ]);

            var_dump($result);
            var_dump('ok');
        });
    }
}

消費(fèi)者

namespace App\Consumer;

use EasySwoole\Component\Process\AbstractProcess;
use EasySwoole\Kafka\Config\ConsumerConfig;
use EasySwoole\Kafka\Kafka;

class Process extends AbstractProcess
{
    protected function run($arg)
    {
        go(function () {
            $config = new ConsumerConfig();
            $config->setRefreshIntervalMs(1000);
            $config->setMetadataBrokerList('127.0.0.1:9092,127.0.0.1:9093');
            $config->setBrokerVersion('0.9.0');
            $config->setGroupId('test');

            $config->setTopics(['test']);
            $config->setOffsetReset('earliest');

            $kafka = new Kafka($config);
            // 設(shè)置消費(fèi)回調(diào)
            $func = function ($topic, $partition, $message) {
                var_dump($topic);
                var_dump($partition);
                var_dump($message);
            };
            $kafka->consumer()->subscribe($func);
        });
    }
}

附贈(zèng)

1. Kafka 集群部署 docker-compose.yml 一份，使用方式如下
   1. 保證2181,9092,9093,9000端口未被占用（占用后可以修改compose文件中的端口號(hào)）
   2. 根目錄下，docker-compose up -d
   3. 訪問(wèn)localhost:9000，可以查看kafka集群狀態(tài)。

https://github.com/easy-swoole/kafka/blob/master/docker-compose.yml