什么是Kafka

Apache Kafka是一个分布式发布 - 订阅消息系统和一个强大的队列，可以处理大量的数据，并使您能够将消息从一个端点传递到另一个端点。 Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上，并在群集内复制以防止数据丢失。 Kafka构建在ZooKeeper同步服务之上。 它与Apache Storm和Spark非常好地集成，用于实时流式数据分析。

工作流程

image

image

发布 - 订阅消息的工作流程
以下是Pub-Sub消息的逐步工作流程 -

1.生产者定期向主题发送消息。

2.Kafka代理存储为该特定主题配置的分区中的所有消息。 它确保消息在分区之间平等共享。 如果生产者发送两个消息并且有两个分区，Kafka将在第一分区中存储一个消息，在第二分区中存储第二消息。

3.消费者订阅特定主题。

4.一旦消费者订阅主题，Kafka将向消费者提供主题的当前偏移，并且还将偏移保存在Zookeeper系综中。

5.消费者将定期请求Kafka(如100 Ms)新消息。

6.一旦Kafka收到来自生产者的消息，它将这些消息转发给消费者。

7.消费者将收到消息并进行处理。

8.一旦消息被处理，消费者将向Kafka代理发送确认。

9.一旦Kafka收到确认，它将偏移更改为新值，并在Zookeeper中更新它。 由于偏移在Zookeeper中维护，消费者可以正确地读取下一封邮件，即使在服务器暴力期间。

10.以上流程将重复，直到消费者停止请求。

11.消费者可以随时回退/跳到所需的主题偏移量，并阅读所有后续消息。

队列消息/用户组的工作流

1.在队列消息传递系统而不是单个消费者中，具有相同组ID 的一组消费者将订阅主题。 简单来说，订阅具有相同 Group ID 的主题的消费者被认为是单个组，并且消息在它们之间共享。 让我们检查这个系统的实际工作流程。

2.生产者以固定间隔向某个主题发送消息。

3.Kafka存储在为该特定主题配置的分区中的所有消息，类似于前面的方案。

4.单个消费者订阅特定主题，假设 Topic-01 为 Group ID 为 Group-1 。

5.Kafka以与发布 - 订阅消息相同的方式与消费者交互，直到新消费者以相同的组ID 订阅相同主题 Topic-01 1 。

6.一旦新消费者到达，Kafka将其操作切换到共享模式，并在两个消费者之间共享数据。 此共享将继续，直到用户数达到为该特定主题配置的分区数。

7.一旦消费者的数量超过分区的数量，新消费者将不会接收任何进一步的消息，直到现有消费者取消订阅任何一个消费者。 出现这种情况是因为Kafka中的每个消费者将被分配至少一个分区，并且一旦所有分区被分配给现有消费者，新消费者将必须等待。

此功能也称为使用者组。 同样，Kafka将以非常简单和高效的方式提供两个系统中最好的。

安装

在kafka官网 http://kafka.apache.org/downloads 下载到最新的kafka安装包
解压到/usr/local/kafka 目录 就完成安装了

配置

server.properties 大概31行的位置

#192.168.233.139是我机器的IP地址 默认是localhost
listeners=PLAINTEXT://192.168.233.139:9092

启动zookeeper

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

启动kafka

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

创建topic

# localhost:2181 是zookeeper监听地址端口
#test是topic名
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

查看topic

bin/kafka-topics.sh --list --zookeeper localhost:2181

创建生产者

　bin/kafka-console-producer.sh --broker-list 192.168.233.139:9092 --topic test

创建消费者

bin/kafka-console-consumer.sh --bootstrap-server 192.168.233.139:9092 --topic test --from-beginning

PHP kafka客户端

安装librdkafka

git clone https://github.com/edenhill/librdkafka.git
./configure
make
sudo make install

安装php-rdkafka

sudo pecl install rdkafka
#Add the following line to your php.ini file:
extension=rdkafka.so
#重启服务器，查看phpinfo，即安装好了rdkafka拓展

生产者

ini_set('display_errors', 1);
error_reporting(-1);
date_default_timezone_set('PRC');

$conf = new RdKafka\Conf();
$conf->set('socket.timeout.ms', 50); // or socket.blocking.max.ms, depending on librdkafka version
//$conf->set('log_level', (string) LOG_DEBUG);
//$conf->set('debug', 'all');


if (function_exists('pcntl_sigprocmask')) {
    pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
    $conf->set('internal.termination.signal', SIGIO);
} else {
    $conf->set('queue.buffering.max.ms', 1);
}

$rk = new RdKafka\Producer($conf);
//$rk->addBrokers("10.0.0.1:9092,10.0.0.2:9092");
$rk->addBrokers("192.168.233.139:9092");
$topic = $rk->newTopic("test");

while(true)
{
    for($i=0;$i<10000;$i++)
    {
        $topic->produce(RD_KAFKA_PARTITION_UA, 0, "Message {$i}");
        $rk->poll(0);
        while ($rk->getOutQLen() > 0) {
            $rk->poll(1);
        }
        echo "send:".$i."\n";
        sleep(1);
    }

}

消费者

ini_set('display_errors', 1);
error_reporting(-1);

$conf = new RdKafka\Conf();

##如果consumeStart 使用RD_KAFKA_OFFSET_STORED 则需要添加此行
$conf->set('group.id', 'consumerGroup1');
if (function_exists('pcntl_sigprocmask')) {
    pcntl_sigprocmask(SIG_BLOCK, array(SIGIO));
    $conf->set('internal.termination.signal', SIGIO);
} else {
    $conf->set('queue.buffering.max.ms', 1);
}

//$conf->set('log_level', (string) LOG_DEBUG);
//$conf->set('debug', 'all');
//$conf->set("debug", "topic,fetch");
$rk = new RdKafka\Consumer($conf);
$rk->addBrokers("192.168.233.139");

$topic = $rk->newTopic("test");
// The first argument is the partition to consume from.
// The second argument is the offset at which to start consumption. Valid values
// are: RD_KAFKA_OFFSET_BEGINNING, RD_KAFKA_OFFSET_END, RD_KAFKA_OFFSET_STORED.
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);


while (true) {
    // The first argument is the partition (again).
    // The second argument is the timeout.
    $msg = $topic->consume(0, 1000);
    if (null === $msg || $msg->err === RD_KAFKA_RESP_ERR__PARTITION_EOF) {
        // Constant check required by librdkafka 0.11.6. Newer librdkafka versions will return NULL instead.
        continue;
    } elseif ($msg->err) {
        echo $msg->errstr(), "\n";
        break;
    } else {
        echo $msg->payload, "\n";
    }
}

image.png

image.png

参考资料

https://www.w3cschool.cn/apache_kafka/apache_kafka_fundamentals.html
https://github.com/arnaud-lb/php-rdkafka#using-stored-offsets