kafka

Kafka是一种高吞吐量的分布式发布订阅消息系统，有如下特性： 通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。 高吞吐量：即使是非常普通的硬件Kafka也可以支持每秒数百万的消息。 支持通过Kafka服务器和消费机集群来分区消息。 支持Hadoop并行数据加载。

本项目代码参考自 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

仓库地址

easyswoole/kafka

基本使用

注册kafka服务

  1. namespace EasySwoole\EasySwoole;
  3. use App\Producer\Process as ProducerProcess;
  4. use App\Consumer\Process as ConsumerProcess;
  5. use EasySwoole\EasySwoole\Swoole\EventRegister;
  6. use EasySwoole\EasySwoole\AbstractInterface\Event;
  7. use EasySwoole\Http\Request;
  8. use EasySwoole\Http\Response;
  10. class EasySwooleEvent implements Event
  11. {
  13.     public static function initialize()
  14.     {
  15.         // TODO: Implement initialize() method.
  16.         date_default_timezone_set('Asia/Shanghai');
  17.     }
  19.     public static function mainServerCreate(EventRegister $register)
  20.     {
  21.         // TODO: Implement mainServerCreate() method.
  22.         // 生产者
  23.         \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new ProducerProcess());
  24.         // 消费者
  25.         \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new ConsumerProcess());
  26.     }
  27. }

生产者

  1. namespace App\Producer;
  3. use EasySwoole\Component\Process\AbstractProcess;
  4. use EasySwoole\Kafka\Config\ProducerConfig;
  5. use EasySwoole\Kafka\Kafka;
  7. class Process extends AbstractProcess
  8. {
  9.     protected function run($arg)
  10.     {
  11.         go(function () {
  12.             $config = new ProducerConfig();
  13.             $config->setMetadataBrokerList('127.0.0.1:9092,127.0.0.1:9093');
  14.             $config->setBrokerVersion('0.9.0');
  15.             $config->setRequiredAck(1);
  17.             $kafka = new Kafka($config);
  18.             $result = $kafka->producer()->send([
  19.                 [
  20.                     'topic' => 'test',
  21.                     'value' => 'message--',
  22.                     'key'   => 'key--',
  23.                 ],
  24.             ]);
  26.             var_dump($result);
  27.             var_dump('ok');
  28.         });
  29.     }
  30. }

消费者

  1. namespace App\Consumer;
  3. use EasySwoole\Component\Process\AbstractProcess;
  4. use EasySwoole\Kafka\Config\ConsumerConfig;
  5. use EasySwoole\Kafka\Kafka;
  7. class Process extends AbstractProcess
  8. {
  9.     protected function run($arg)
  10.     {
  11.         go(function () {
  12.             $config = new ConsumerConfig();
  13.             $config->setRefreshIntervalMs(1000);
  14.             $config->setMetadataBrokerList('127.0.0.1:9092,127.0.0.1:9093');
  15.             $config->setBrokerVersion('0.9.0');
  16.             $config->setGroupId('test');
  18.             $config->setTopics(['test']);
  19.             $config->setOffsetReset('earliest');
  21.             $kafka = new Kafka($config);
  22.             // 设置消费回调
  23.             $func = function ($topic, $partition, $message) {
  24.                 var_dump($topic);
  25.                 var_dump($partition);
  26.                 var_dump($message);
  27.             };
  28.             $kafka->consumer()->subscribe($func);
  29.         });
  30.     }
  31. }

附赠

  1. Kafka 集群部署 docker-compose.yml 一份，使用方式如下
    1. 保证2181,9092,9093,9000端口未被占用（占用后可以修改compose文件中的端口号）
    2. 根目录下，docker-compose up -d
    3. 访问localhost:9000，可以查看kafka集群状态。

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