日志:117 评论:13
北京, 西城
【Kafka】Kafka学习笔记

介绍

介绍

Apache Kafka是一个开源消息系统项目,该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。

安装

准备

安装依赖

yum -y install wget

建立安装目录

mkdir -p /tongfu.net/env

建立软件包目录

mkdir /packages
cd /packages

下载

kafka

http://kafka.apache.org/downloads

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.1/kafka_2.12-2.2.1.tgz

JDK

必须通过网页下载,需要勾选“接受协议”,所以不能用wget直接下载了!

网页地址:https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

安装

安装JDK

tar -xzvf jdk-8u192-linux-x64.tar.gz -C /tongfu.net/env/

安装Kafka

tar -xzvf kafka_2.12-2.2.1.tgz -C /tongfu.net/env/

配置

配置JDK

echo "export JAVA_HOME=/tongfu.net/env/jdk1.8.0_192" >> /etc/profile
echo "export CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar" >> /etc/profile
source /etc/profile

配置Kafka

进入kafka主目录

cd /tongfu.net/env/kafka_2.12-2.2.1

打开kafka配置文件

vi /tongfu.net/env/kafka_2.12-2.2.1/config/server.properties

设置日志路径

log.dirs=/tongfu.net/env/kafka_2.12-2.2.1/kafka-logs

设置JVM参数

设置zookeeper的JVM参数

[root@tongfunet]# vi ./bin/zookeeper-server-start.sh
export KAFKA_HEAP_OPTS="-Xmx128M -Xms64M"

设置kafka的JVM参数

[root@tongfunet]# vi ./bin/kafka-server-start.sh
export KAFKA_HEAP_OPTS="-Xmx256M -Xms128M"

启动

启动zookeeper

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

启动kafka

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

自动启动

编写启动脚本

[root@tongfunet]# cat > /tongfu.net/env/kafka_2.12-2.2.1/bin/kafka-start <<EOF
#!/bin/sh

# env
export PATH

# start
source /etc/profile

cd /tongfu.net/env/kafka_2.12-2.2.1/

./bin/zookeeper-server-start.sh ./config/zookeeper.properties >> ./logs/zookeeper.log &
./bin/kafka-server-start.sh ./config/server.properties >> ./logs/kafka.log &
EOF

编写停止脚本

[root@tongfunet]# cat > /tongfu.net/env/kafka_2.12-2.2.1/bin/kafka-quit <<EOF
#!/bin/sh

# env
export PATH

# quit
source /etc/profile

cd /tongfu.net/env/kafka_2.12-2.2.1/

./bin/kafka-server-stop.sh
./bin/zookeeper-server-stop.sh
EOF

授权脚本

chmod 0755 /tongfu.net/env/kafka_2.12-2.2.1/bin/kafka-start
chmod 0755 /tongfu.net/env/kafka_2.12-2.2.1/bin/kafka-quit

添加自动启动脚本

[root@tongfunet]# cat > /lib/systemd/system/kafka.service <<EOF
[Unit]
Description=kafka
After=network.target

[Service]
Type=forking
ExecStart=/tongfu.net/env/kafka_2.12-2.2.1/bin/kafka-start
ExecStop=/tongfu.net/env/kafka_2.12-2.2.1/bin/kafka-quit
PrivateTmp=true

[Install]
WantedBy=multi-user.target
EOF

使用 systemctl 管理 kafka 服务

systemctl enable kafka # 设置自动启动

systemctl start kafka # 启动服务

systemctl stop kafka # 停止服务

systemctl restart kafka # 重启服务

使用

shell测试

创建topic

创建一个名为 test 的 topic

  • --replication-factor:指定副本数量,就是说可以设定一个值,在kafka集群里将一个主题制作多个副本以防止数据丢失。当然,副本数量不能超过集群节点总数。(废话!)

  • --partitions:指定分区数量,就是说可以将一个主题下的每个消息分配到不同的房间里,来自同一分组的多个消费者来获取消息的时候,每个消费者只能拿到一个房间内的消息

  • --topic:主题名称

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

开启消费进程

开启一个新的终端,开启一个来自 localhost:9092 的消费进程来监听名为 test 的 topic

  • --topic:主题名称

  • --from-beginning:从最前面开始提取消息

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

启动生产进程

开启另外一个终端,启动一个生产进程用于推送消息到 localhost:9092 上的名为 test 的 topic

  • --topic:主题名称

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

启动后在终端输入消息后,会在消费进程里看到这条消息

PHP调用

我们使用PHP去操作kafka的方法如下

安装编译环境

yum -y install gcc gcc-c++ make

安装扩展

下载librdkafka

wget -c https://github.com/edenhill/librdkafka/archive/v0.11.0.tar.gz

下载php-rdkafka

wget -c https://github.com/arnaud-lb/php-rdkafka/archive/3.0.4.tar.gz

安装librdkafka

tar -xzvf v0.11.0.tar.gz
cd librdkafka-0.11.0/
./configure --prefix=/tongfu.net/env/librdkafka/
make
make install
cd ..

安装php-rdkafka

tar -xzvf 3.0.4.tar.gz
cd php-rdkafka-3.0.4/
/tongfu.net/env/php-7.2.8/bin/phpize
./configure \
--with-php-config=/tongfu.net/env/php-7.2.8/bin/php-config \
--with-rdkafka=/tongfu.net/env/librdkafka/
make
make install
cd ..

配置php.ini

extension=rdkafka

低级消费者概念

低级消费者模式就是一个人接了一个项目,所有事情都要自己一点点做。

如果当天没有做完,我会记录做到第几件了,然后第二天来了继续做下面的工作。

低级消费者代码

低级消费者示例代码 

  • 当 $message 为 null 或者 $message->err 为到底了,都表示没有新消息

  • 强烈建议本地保存 offset(偏移),我做了多少我自己记着,放心!

$groupID = "lowLevel";
$topicName = "TestMP";
$partitionID = 0;

$conf = new RdKafka\Conf();
$conf->setErrorCb(function($kafka, $err, $reason){
    var_dump($err);
    var_dump($reason);
});
$conf->set('group.id', $groupID);

$consumer = new RdKafka\Consumer($conf);
$consumer->addBrokers("server_kafka:9092");

$topicConf = new RdKafka\TopicConf();
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', __DIR__);
$topicConf->set('auto.commit.enable', 1);
$topicConf->set('auto.commit.interval.ms', 10);
$topicConf->set('auto.offset.reset', 'smallest');

$topic = $consumer->newTopic($topicName, $topicConf);
$topic->consumeStart($partitionID, RD_KAFKA_OFFSET_STORED);

var_dump("Starting...");
while (1) {
    // try consumer record
    var_dump("Waiting...");
    $message = $topic->consume($partitionID, 120*1000);
    if($message == null){
         continue;
    }
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            var_dump($message->payload);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            var_dump("No more messages");
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            var_dump("Timeout");
            break;
        default:
            throw new Exception($message->errstr(), $message->err);
            break;
    }
}

高级消费者概念

高级消费者模式就是团队做项目,要做的事情列出来了,你接一个,我接一个,各自做完手头的工作后再去接下一个。

分配工作的是 Kafka 的 Rebalance 来控制,逻辑也非常简单。

它先把项目需要做的事情分开多个袋子里,然后每个袋子上写一个编号。

当员工来接活的时候,会先看看有没有没人负责的袋子,如果有就告诉员工你的编号是这个,这个袋子里的事情都是你的。

如果全部袋子都有人负责了,那就告诉员工暂时没事干,你先等等吧。

注意:消费者能不能拿到消息,完全看是不是可以得到分配到分区,而往往是这个地方会需要等很久~~

高级消费者代码

高级消费者示例代码

$groupID = "highLevel";
$topicName = "TestMP";

$conf = new RdKafka\Conf();
$conf->setErrorCb(function($kafka, $err, $reason){
    var_dump($err);
    var_dump($reason);
});
$conf->setRebalanceCb(function (RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
    switch ($err) {
        case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
            foreach($partitions as $partition){
                var_dump("Assign: ". $partition->getPartition());
            }
            $kafka->assign($partitions);
            break;

        case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
            foreach($partitions as $partition){
                var_dump("Revoke: ". $partition->getPartition());
            }
            $kafka->assign(NULL);
            break;

        default:
            throw new \Exception($err);
    }
});
$conf->set('group.id', $groupID);
$conf->set('metadata.broker.list', "server_kafka:9092");

$topicConf = new RdKafka\TopicConf();
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', __DIR__);
$topicConf->set('auto.commit.enable', 1);
$topicConf->set('auto.commit.interval.ms', 10);
$topicConf->set('auto.offset.reset', 'smallest');

$conf->setDefaultTopicConf($topicConf);

$consumer = new RdKafka\KafkaConsumer($conf);
$consumer->subscribe(array($topicName));

var_dump("Starting...");
while (1) {
    // try consumer record
    var_dump("Waiting...");
    $message = $consumer->consume(120*1000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            var_dump($message->payload);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            var_dump("No more messages");
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            var_dump("Timeout");
            break;
        default:
            throw new Exception($message->errstr(), $message->err);
            break;
    }
}

生产者示例代码

生产者示例代码

  • 使用 setErrorCb 回调检查程序是否出错

  • 使用 setDrMsgCb 回调检查消息是否成功推送

  • 设置 message.timeout 避免 produce 阻塞

$conf = new RdKafka\Conf();
$conf->setErrorCb(function($kafka, $err, $reason){
    var_dump($err);
    var_dump($reason);
});
$conf->setDrMsgCb(function($kafka, $message){
    if($message->err) var_dump("Delivery fail: ");
    else var_dump("Delivery success: ");
    var_dump($message);
});

$rk = new RdKafka\Producer($conf);
$rk->addBrokers("server_kafka:9092");

$topicConf = new RdKafka\TopicConf();
$topicConf->set("message.timeout.ms", 3000);

$topic = $rk->newTopic("test", $topicConf);
$topic->produce(RD_KAFKA_PARTITION_UA, 0, date("Y-m-d H:i:s"));

官方在线文档

Rdkafka扩展提供了在线文档,可以帮助我们编写代码的时候查阅

http://arnaud.le-blanc.net/php-rdkafka/phpdoc/book.rdkafka.html

分区

分区

一个topic 可以配置几个partition,produce发送的消息分发到不同的partition中,consumer接受数据的时候是按照group来接受,kafka确保每个partition只能同一个group中的同一个consumer消费,如果想要重复消费,那么需要其他的组来消费。

Zookeerper中保存这每个topic下的每个partition在每个group中消费的offset 

新版kafka把这个offsert保存到了一个__consumer_offsert的topic下 
这个__consumer_offsert 有50个分区,通过将group的id哈希值%50的值来确定要保存到那一个分区.  这样也是为了考虑到zookeeper不擅长大量读写的原因。

所以,如果要一个group用几个consumer来同时读取的话,需要多线程来读取,一个线程相当于一个consumer实例。当consumer的数量大于分区的数量的时候,有的consumer线程会读取不到数据。 

假设一个topic test 被groupA消费了,现在启动另外一个新的groupB来消费test,默认test-groupB的offset不是0,而是没有新建立,除非当test有数据的时候,groupB会收到该数据,该条数据也是第一条数据,groupB的offset也是刚初始化的ofsert, 除非用显式的用–from-beginnging 来获取从0开始数据

备注

文档中文注解

server.properties

#broker的全局唯一编号,不能重复

broker.id=0

 

#用来监听链接的端口,producer或consumer将在此端口建立连接

port=9092

 

#处理网络请求的线程数量

num.network.threads=3

 

#用来处理磁盘IO的线程数量

num.io.threads=8

 

#发送套接字的缓冲区大小

socket.send.buffer.bytes=102400

 

#接受套接字的缓冲区大小

socket.receive.buffer.bytes=102400

 

#请求套接字的缓冲区大小

socket.request.max.bytes=104857600

 

tfart_kafka消息存放的路径

log.dirs=/home/servers-kafka/logs/kafka

 

tfart_topic在当前broker上的分片个数

num.partitions=2

 

#用来恢复和清理data下数据的线程数量

num.recovery.threads.per.data.dir=1

 

tfart_segment文件保留的最长时间,超时将被删除

log.retention.hours=168

 

#滚动生成新的segment文件的最大时间

log.roll.hours=168

 

#日志文件中每个segment的大小,默认为1G

log.segment.bytes=1073741824

 

#周期性检查文件大小的时间

log.retention.check.interval.ms=300000

 

#日志清理是否打开

log.cleaner.enable=true

 

tfart_broker需要使用zookeeper保存meta数据

zookeeper.connect=hadoop02:2181,hadoop03:2181,hadoop04:2181

 

tfart_zookeeper链接超时时间

zookeeper.connection.timeout.ms=6000

 

tfart_partion buffer中,消息的条数达到阈值,将触发flush到磁盘

log.flush.interval.messages=10000

 

#消息buffer的时间,达到阈值,将触发flush到磁盘

log.flush.interval.ms=3000

 

#删除topic需要server.properties中设置delete.topic.enable=true否则只是标记删除

delete.topic.enable=true

 

#此处的host.name为本机IP(重要),如果不改,则客户端会抛出:Producerconnection to localhost:9092 unsuccessful 错误!

host.name=hadoop02

 

2、修改producer.properties

 

 #指定kafka节点列表,用于获取metadata,不必全部指定

metadata.broker.list=hadoop02:9092,hadoop03:9092

 

# 指定分区处理类。默认kafka.producer.DefaultPartitioner,表通过key哈希到对应分区

tfart_partitioner.class=kafka.producer.DefaultPartitioner

 

# 是否压缩,默认0表示不压缩,1表示用gzip压缩,2表示用snappy压缩。压缩后消息中会有头来指明消息压缩类型,故在消费者端消息解压是透明的无需指定。

compression.codec=none

 

# 指定序列化处理类

serializer.class=kafka.serializer.DefaultEncoder

 

# 如果要压缩消息,这里指定哪些topic要压缩消息,默认empty,表示不压缩。

tfart_compressed.topics=

 

# 设置发送数据是否需要服务端的反馈,有三个值0,1,-1

# 0: producer不会等待broker发送ack

# 1: 当leader接收到消息之后发送ack

# -1: 当所有的follower都同步消息成功后发送ack.

request.required.acks=0

 

#在向producer发送ack之前,broker允许等待的最大时间 ,如果超时,broker将会向producer发送一个error ACK.意味着上一次消息因为某种原因未能成功(比如follower未能同步成功)

request.timeout.ms=10000

 

# 同步还是异步发送消息,默认“sync”表同步,"async"表异步。异步可以提高发送吞吐量,

也意味着消息将会在本地buffer中,并适时批量发送,但是也可能导致丢失未发送过去的消息

producer.type=sync

 

# 在async模式下,当message被缓存的时间超过此值后,将会批量发送给broker,默认为5000ms

# 此值和batch.num.messages协同工作.

queue.buffering.max.ms = 5000

 

# 在async模式下,producer端允许buffer的最大消息量

# 无论如何,producer都无法尽快的将消息发送给broker,从而导致消息在producer端大量沉积

# 此时,如果消息的条数达到阀值,将会导致producer端阻塞或者消息被抛弃,默认为10000

queue.buffering.max.messages=20000

 

# 如果是异步,指定每次批量发送数据量,默认为200

batch.num.messages=500

 

# 当消息在producer端沉积的条数达到"queue.buffering.max.meesages"后

# 阻塞一定时间后,队列仍然没有enqueue(producer仍然没有发送出任何消息)

# 此时producer可以继续阻塞或者将消息抛弃,此timeout值用于控制"阻塞"的时间

# -1: 无阻塞超时限制,消息不会被抛弃

# 0:立即清空队列,消息被抛弃

queue.enqueue.timeout.ms=-1

 

 

# 当producer接收到error ACK,或者没有接收到ACK时,允许消息重发的次数

# 因为broker并没有完整的机制来避免消息重复,所以当网络异常时(比如ACK丢失)

# 有可能导致broker接收到重复的消息,默认值为3.

message.send.max.retries=3

 

# producer刷新topicmetada的时间间隔,producer需要知道partitionleader的位置,以及当前topic的情况

# 因此producer需要一个机制来获取最新的metadata,当producer遇到特定错误时,将会立即刷新

#(比如topic失效,partition丢失,leader失效等),此外也可以通过此参数来配置额外的刷新机制,默认值600000

topic.metadata.refresh.interval.ms=60000

 

3、修改consumer.properties

 

# zookeeper连接服务器地址

zookeeper.connect=hadoop02:2181,hadoop03:2181,hadoop04:2181

 

# zookeeper的session过期时间,默认5000ms,用于检测消费者是否挂掉

zookeeper.session.timeout.ms=5000

 

#当消费者挂掉,其他消费者要等该指定时间才能检查到并且触发重新负载均衡

zookeeper.connection.timeout.ms=10000

 

# 指定多久消费者更新offset到zookeeper中。注意offset更新时基于time而不是每次获得的消息。一旦在更新zookeeper发生异常并重启,将可能拿到已拿到过的消息

zookeeper.sync.time.ms=2000

 

#指定消费组

group.id=xxx

 

# 当consumer消费一定量的消息之后,将会自动向zookeeper提交offset信息

# 注意offset信息并不是每消费一次消息就向zk提交一次,而是现在本地保存(内存),并定期提交,默认为true

auto.commit.enable=true

 

# 自动更新时间。默认60 * 1000

auto.commit.interval.ms=1000

 

# 当前consumer的标识,可以设定,也可以有系统生成,主要用来跟踪消息消费情况,便于观察

conusmer.id=xxx

 

# 消费者客户端编号,用于区分不同客户端,默认客户端程序自动产生

client.id=xxxx

 

# 最大取多少块缓存到消费者(默认10)

queued.max.message.chunks=50

 

# 当有新的consumer加入到group时,将会reblance,此后将会有partitions的消费端迁移到新  的consumer上,如果一个consumer获得了某个partition的消费权限,那么它将会向zk注册"Partition Owner registry"节点信息,但是有可能此时旧的consumer尚没有释放此节点, 此值用于控制,注册节点的重试次数.

rebalance.max.retries=5

 

# 获取消息的最大尺寸,broker不会像consumer输出大于此值的消息chunk 每次feth将得到多条消息,此值为总大小,提升此值,将会消耗更多的consumer端内存

fetch.min.bytes=6553600

 

# 当消息的尺寸不足时,server阻塞的时间,如果超时,消息将立即发送给consumer

fetch.wait.max.ms=5000

socket.receive.buffer.bytes=655360

 

# 如果zookeeper没有offset值或offset值超出范围。那么就给个初始的offset。有smallest、largest、anything可选,分别表示给当前最小的offset、当前最大的offset、抛异常。默认largest

auto.offset.reset=smallest

 

# 指定序列化处理类

derializer.class=kafka.serializer.DefaultDecoder