1 php拓展所在位置

可以到php.ini中查看ext的存放位置

php/ext文件中存放的后缀为dll（动态链接库）的文件为php的拓展组件。

2 查看phpinfo

  1. php -i // 输出php信息，等同于phpinfo
  2. php -r 'phpinfo();' // 命令行执行code

使用phpinfo()函数查看PHP的版本信息，这会决定扩展文件版本

php安装拓展 - 图1

其中，第一个是VC的版本为11，第二个是64位的，第三个参数api版本号，安装的组件是TS（即线程安全）的，VC版本为11

3 安装redis

3.1 找对应的redis.dll和igbinary.dll

igbinary是安装redis的前提即依赖

在mac上也是需要igbinary的，linux这类的操作系统可以通过brew安装php,自带pecl，使用pecl安装拓展，它是php第三方拓展库

dll即动态链接库，是C语言编写的，已经编译好之后的库。而在linux上，多采用二进制源码编译的方式，得到so后缀的文件

http://windows.php.net/downloads/pecl/releases/redis/2.2.7/

php安装拓展 - 图2

http://windows.php.net/downloads/pecl/releases/igbinary/2.0.1/

php安装拓展 - 图3

3.2 修改php.ini

  1. [Redis]
  2. ;php_redis
  3. extension=php_igbinary.dll
  4. extension=php_redis.dll

注意：extension=php_igbinary.dll要放在extension=php_redis.dll前，否则不会生效。

在mac中，还需要重启php-fpm，有时候它会自动平滑重启，有时候需要自己重启。

  1. ps aux|grep php-fpm // 查看进程号
  2. kill -USR2 php-fpm.pid的绝对路径或者PID号
  3. php --ri redis // 查看拓展配置信息以便确定重启成功
  4. php -m // 查看所有的拓展

3.3 纠错

出现

  1.  Unable to load dynamic library 'f:\php\ext\php_redis.dll' - 找不到指定的模块。

或者

  1. Gracefully shutting down php-fpm . done
  2. Starting php-fpm [16-Jan-2014 12:46:26] NOTICE: PHP message: PHP Warning:  PHP Startup: memcache: Unable to initialize module
  3. Module compiled with module API=20090626
  4. PHP    compiled with module API=20100525
  5. These options need to match
  6.  in Unknown on line 0
  7.  done

即可能是下载的时候疏忽了，没用下载到对应php版本的拓展，即版本不匹配，重新下载吧。

4 安装kafka拓展

安装kafka集群：https://www.yuque.com/u21438008/nd8ycn/urfi5m
php手册介绍：php需在7以上。https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/rdkafka.requirements.html
函数：https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/index.html
安装方式：
https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/rdkafka.installation.windows.html
image.png

  1. <?php
  3. class KafkaHelper {
  4.     private $topicName = '';
  5.     private $ipAddress = '';
  6.     private $groupName = '';
  8.     // 生产者配置
  9.     private $conf = null;
  11.     // 生产者topic配置（感觉抽离出来是多topic）
  12.     private $topicConf = null;
  14.     // 生产者
  15.     private $producer = null;
  17.     // 发送操作对象（感觉抽离出来是多topic）
  18.     private $producerTopic = null;
  20.     // 
  21.     // 消费者配置
  22.     private $consumerConf = null;
  24.     // 消费者topic 配置
  25.     private $consumerTopicConf = null;
  27.     // 消费者
  28.     private $consumer = null;
  30.     // 消费操作对象
  31.     private $consumerTopic = null;
  33.     // 初始话设置相关配置信息
  34.     public function __construct($topicName = '', $ipAddress = '', $groupName = '')
  35.     {
  36.         $this->topicName = $topicName;
  37.         echo "topic_name:" . $topicName . "\n";
  38.         $this->ipAddress = $ipAddress;
  39.         echo "ipaddress:" . $ipAddress . "\n";
  40.         $this->groupName = $groupName;
  41.         echo "groupName:" . $groupName . "\n";
  42.     }
  44.     // https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
  45.     public function setConf($conf = null)
  46.     {
  47.         if (!$conf) {
  48.             $conf = new \RdKafka\Conf();
  49.             $conf->set('log_level', (string) LOG_DEBUG);
  50.             $conf->set('metadata.broker.list', $this->ipAddress);
  51.             $conf->setDrMsgCb(function ($kafka, $msg) {
  52.                 if ($msg->err) {
  53.                     // message permanently failed to be delivered
  54.                     echo "produce failed:" . json_encode($msg) . "\n";
  55.                 } else {
  56.                     // message successfully delivered
  57.                     echo "produce succeed:" . json_encode($msg) . "\n";
  58.                 }
  59.             });
  60.             $this->conf = $conf;
  61.         } else {
  62.             if (!($conf instanceof \RdKafka\Conf)) {
  63.                 throw new \Exception('传入的配置类应为: \Rdkafka\conf');
  64.             }
  65.             $this->conf = $conf;
  66.         }
  68.         return $this;
  69.     }
  71.     public function setTopicConf($topicConf = null)
  72.     {
  73.         if (!$topicConf) {
  74.             $topicConf = new \Rdkafka\TopicConf();
  75.             $topicConf->set('request.required.acks', 0);
  76.             $this->topicConf = $topicConf;
  77.         } else {
  78.             if (!($topicConf instanceof \Rdkafka\TopicConf)) {
  79.                 throw new \Exception('传入的配置类应为: \Rdkafka\TopicConf');
  80.             }
  81.             $this->topicConf = $topicConf;
  82.         }
  83.         return $this;
  84.     }
  86.     public function createProducer()
  87.     {
  88.         $this->producer = new \RdKafka\Producer($this->conf);
  89.         // $this->producer->addBrokers($this->ipAddress);
  90.         $this->producerTopic = $this->producer->newTopic($this->topicName, $this->topicConf);
  91.         return $this;
  92.     }
  94.     // 二维数组
  95.     public function send($data = [])
  96.     {
  97.         if (empty($data)) {
  98.             return '数据不能为空';
  99.         }
  100.         echo "sending:". json_encode($data) ."\n";
  101.         try {
  102.             foreach ($data as $key => $val) {
  103.                 echo "$key : ". json_encode($val) ."\n";
  104.                 $this->producerTopic->produce(RD_KAFKA_PARTITION_UA, 0, json_encode($val, JSON_UNESCAPED_UNICODE));
  105. //                LogHelper::log('发送kafka数据：'.json_encode($val, JSON_UNESCAPED_UNICODE), 'kafka_producer');
  106.             }
  107.             $times = 0;
  108.             while ($this->producer->getOutQLen() > 0) {
  109. //                LogHelper::log($this->rk->getOutQLen(), 'kafka_producer');
  110.                 if ($times++ > 10) {
  111. //                    LogHelper::log('投递kafka超时：','kafka_producer');
  112.                     throw new \Exception('消息超时');
  113.                 }
  114.                 $this->producer->poll(300);
  115.             }
  116.             echo "end\n";
  117.             return true;
  118.         } catch (\Exception $e) {
  119.             echo 'errormsg:' . $e->getMessage() . "\n";
  120. //            LogHelper::log('投递kafka异常：'.$e->getMessage(),'kafka_producer');
  121.             return false;
  122.         }
  123.     }
  125.     // create consumerConf. You should call ConsumerTopicConf() first.
  126.     public function setConsumerConf($consumerConf = null)
  127.     {
  128.         echo "set consumer conf\n";
  129.         if (!$consumerConf) {
  130.             $consumerConf = new \RdKafka\Conf();
  131.             // 配置groud.id 具有相同 group.id 的consumer 将会处理不同分区的消息，所以同一个组内的消费者数量如果订阅了一个topic， 那么消费者进程的数量多于 多于这个topic 分区的数量是没有意义的。
  132.             $consumerConf->set('group.id', $this->groupName);
  133.             //添加 kafka集群服务器地址
  134.             $consumerConf->set('metadata.broker.list', $this->ipAddress);
  135.             $consumerConf->setDefaultTopicConf($this->consumerTopicConf);
  136.             // 消费成功的回调
  137.             $consumerConf->setConsumeCb(function ($msg) {
  138.                 echo "consume succeed:". json_encode($msg) ."\n";
  139.             });
  140.             $this->consumerConf = $consumerConf;
  141.         } else {
  142.             if (!($consumerConf instanceof \RdKafka\Conf)) {
  143.                 throw new \Exception('传入的配置类应为: \Rdkafka\Conf');
  144.             }
  145.             $this->consumerConf = $consumerConf;
  146.         }
  148.         return $this;
  149.     }
  151.     // 设置消ConsumerTopicConf，无需其他配合
  152.     public function setConsumerTopicConf($consumerTopicConf = null)
  153.     {
  154.         echo "set consumer topic conf\n";
  155.         if (!$consumerTopicConf) {
  156.             $consumerTopicConf = new \RdKafka\TopicConf();
  157.             ////当没有初始偏移量时，从哪里开始读取 'smallest': start from the beginning    'latest' 读取最新的数据
  158.             $consumerTopicConf->set('auto.offset.reset', 'smallest');
  159.             //设置自动记录上次读取到的位置
  160.             //$topicConf->set('auto.commit.enable', true);
  161.             // Set the configuration to use for subscribed/assigned topics
  162.             $this->consumerTopicConf = $consumerTopicConf;
  163.         } else {
  164.             if (!($consumerTopicConf instanceof \Rdkafka\TopicConf)) {
  165.                 throw new \Exception('传入的配置类应为: \Rdkafka\TopicConf');
  166.             }
  167.             $this->consumerTopicConf = $consumerTopicConf;
  168.         }
  169.         return $this;
  170.     }
  172.     public function createConsumer()
  173.     {
  174.         echo "create consumer \n";
  175. //        使用RdkafkaConsumer()貌似就能自动消费所有Partition;Consumer()就需要指定哪个Partition，然后其他的Partition消费不到
  176.         $consumer = new \RdKafka\KafkaConsumer($this->consumerConf);
  177. //        $consumer = new \RdKafka\Consumer($this->consumerConf);
  178.         $this->consumer = $consumer;
  179.         // 让消费者订阅log 主题
  180.         $consumer->subscribe([$this->topicName]);
  181. //        $this->consumTopic = $this->consumer->newTopic($this->topicName, $this->consumerTopicConf);
  182.         return $this;
  183.     }
  185.     public function startConsume()
  186.     {
  187.         echo "start consumer---------" . "\n";
  188. //        $consumerTopic = $this->consumTopic;
  189.         $consumer = $this->consumer;
  190.          try {
  191.              //解决 kafka在遇到PHP fatalerrorr不退出的问题。
  192.              $handleShutdown = function () use ($consumer) {
  193.                  $consumer->unsubscribe();
  194.              };
  195.              register_shutdown_function($handleShutdown);
  196.          } catch (\Exception $e) {
  197.              $consumer && $consumer->unsubscribe();
  198.          }
  200.         // 消费
  201. //        $consumerTopic->consumeStart(0, RD_KAFKA_OFFSET_STORED);
  202.         while (true) {
  203.             try {
  204.                 // 原参数1000
  205. //                $message = $consumerTopic->consume(0, 120*10000);
  206.                 $message = $consumer->consume(1000);
  207.                 // 返回消费结果
  208.                 switch ($message->err) {
  209.                     // 正常消费
  210.                     case RD_KAFKA_RESP_ERR_NO_ERROR:
  211. //                        LogHelper::log('message:kafka信息:' . json_encode($message), $this->logFile);
  212. //                        self::stdOutLineF('message:kafka信息:' . json_encode($message));
  213.                         if (!is_null($message)) {
  214.                             $data = json_decode($message->payload,true);
  215.                             $this->handleData($data); // 处理数据
  216.                         }
  217.                         break;
  218.                     case RD_KAFKA_RESP_ERR__PARTITION_EOF:
  219. //                        LogHelper::log($message->err . " | No more messages; will wait for more", $this->logFile);
  220. //                        self::stdOutLineF($message->err . " | No more messages; will wait for more");
  221.                         break;
  222.                     case RD_KAFKA_RESP_ERR__TIMED_OUT:
  223.                         break;
  224.                     default:
  225.                         throw new \Exception($message->errstr(), $message->err);
  226.                         break;
  227.                 }
  228.             } catch (\Exception $e) {
  229. //                self::stdOutLineF("Kafka Stop Error:".$e->getMessage());
  230. //                LogHelper::log('Kafka Stop Error:' . $e->getMessage(), $this->logFile);
  231.                 $consumer->unsubscribe();
  232. //                $consumerTopic->consumeStop(0, RD_KAFKA_OFFSET_STORED);
  233.                 echo "出错了\n";
  234.             }
  235.         }
  236.     }
  238.     // json_decode后的数据
  239.     public function handleData($data)
  240.     {
  241.         echo json_encode($data) . "\n";
  242.     }
  243. }
  245. $ipAddress = '192.168.184.129:9093';
  246. $topicName = 'TestTopic';
  247. $group = 'test';
  248. $kafka = new KafKaHelper($topicName, $ipAddress, $group);
  250. // $data = [['22' => 'test']];
  251. // $kafka->setConf()->setTopicConf()->createProducer()->send($data);
  253. //
  254. $kafka->setConsumerTopicConf()->setConsumerConf()->createConsumer()->startConsume();

image.png