2019-08-27发表2021-10-11更新Kafka3 分钟读完 (大约502个字)

消息中间件Kafka - PHP操作使用Kafka

PHP使用Kafka

我们需要安装libkafka和rdkafka

安装libkafka

下载

去GitHub上克隆下来

git clone https://github.com/edenhill/librdkafka.git
安装

cd librdkafka/

`./configure && make && make install`

安装成功界面没有报错就是安装成功

![14c1ce5d9719d16db1f2de5e9eed9553.png](https://img-blog.csdnimg.cn/20190320100758703.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMzIwMjgx,size_16,color_FFFFFF,t_70)

安装rdkafka

下载

git clone https://github.com/arnaud-lb/php-rdkafka

cd php-rdkafka/
为php安装扩展

在php-rdkafka这个目录下

`phpize`

然后会生成源代码安装的脚本

把php-config的位置改成自己php-config的位置

` ./configure --with-php-config=/usr/local/php/bin/php-config`

编译安装

`make && make install`

成功后会出现一个文件夹

![70c28ed68f37346a84abe96a2fe03aae.png](https://img-blog.csdnimg.cn/20190320100828577.png)

这个位置就是保存的我们刚刚安装的扩展

进入该目录

`cd /usr/local/php/lib/php/extensions/no-debug-non-zts-20170718/`

会发现出现个rdkafka.so文件

![2bb11d41b5e08b42a3fad4fc181eb791.png](https://img-blog.csdnimg.cn/20190320100957658.png)

修改php.ini文件加入  这里的路径就是写自己rdkafka.so文件的路径

`extension=/usr/local/php/lib/php/extensions/no-debug-non-zts-20170718/rdkafka.so `
重启php

php-m

出现rdkafka就是安装成功

![fe40f168197be21fb00db6d06428df24.png](https://img-blog.csdnimg.cn/20190320101008302.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMzIwMjgx,size_16,color_FFFFFF,t_70)

php操作kafka

运行前先开启我们的zookeeper和kafka 上篇文章有如何开启

运行producer
kafka默认端口9092 vim producer.php

 <?php
    $rk = new RdKafka\Producer();
    $rk->setLogLevel(LOG_DEBUG);
    $rk->addBrokers("ip:9092");       
    $topic = $rk->newTopic("test");
    $topic->produce(RD_KAFKA_PARTITION_UA, 0, "要发送的消息");

运行consumer
vim consumer.php

 <?php
    $rk = new RdKafka\Consumer();
    $rk->setLogLevel(LOG_DEBUG);
    $rk->addBrokers("ip");
    $topic = $rk->newTopic("test");
    $topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);
    while(true){
        sleep(1);
        $msg = $topic->consume(0, 1000);
        if ($msg) {
            echo $msg->payload, "\n";
        }          
    }

开启两个窗口一个运行consumer 一个运行producer php consumer.php
php producer.php会发现我们已经简单的会使用kafka了。

2019-08-27发表2021-10-11更新Kafka4 分钟读完 (大约625个字)

消息中间件Kafka - 介绍及安装

Kafka介绍

优势

高吞吐量：非常普通的硬件Kafka也可以支持每秒数百万的消息
支持通过Kafka服务器和消费机集群来区分消息
支持Hadoop并行数据加载

关键概念

Broker：Kafka集群中的一台或多台服务器统称为broker。
Topic：Kafka处理的消息源（feeds of messages）的不同分类。
Partition：Topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。
Message：消息，是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。
Producers：消息和数据生产者，向Kafka的一个topic发布消息的过程叫做producers。
Consumers：消息和数据的消费者，订阅topics并处理其发布的消息的过程叫做consumers。

安装

下载
先安装jdk 然后jdk的安装方式在elasticsearch的安装文章中有，这里就不写了
kafka官网 wget https://www-us.apache.org/dist/kafka/2.1.1/kafka_2.11-2.1.1.tgz 解压 tar -xzvf kafka_2.11-2.1.1.tgz

修改配置文件

cd kafka_2.11-2.1.1/config

zookeeper.properties 是zookeeper的配置文件，默认端口号2181，可不做修改

server.properties 是kafka配置文件，将 zookeeper.connect 这行改为自己的zookeeper地址和端口号

修改完成之后返回kafka主目录

cd ..
运行zookeeper和kafka

运行zookeeper

bin/zookeeper-server-start.sh config/zookeeper.properties

不要关闭此窗口再开一个新窗口重新进入kafka目录

运行kafka

bin/kafka-server-start.sh config/server.properties
运行producer和consumer

跟上步操作一样不要关闭窗口重新开重新进入kafka目录

创建一个topic为test

把ip和port改为自己zookeeper的

bin/kafka-topics.sh --create --zookeeper ip:port --replication-factor 1 --partitions 1 --topic test

运行producer

bin/kafka-console-producer.sh --broker-list ip:port --topic test

跟上步操作一样不要关闭窗口重新开重新进入kafka目录

运行consumer

bin/kafka-console-consumer.sh --bootstrap-server ip:port --topic test --from-beginning

然后在producer发送信息会发现 consumer的窗口会出现你发送的消息