Kafka 配置

Kafka的消息机制

在 Kafka 中 Topic 是一个存储消息的逻辑概念，可以认为是一个消息集合。每条消息发送到 Kafka 集群的消息都有一个类别。物理上来说，不同的 Topic 的消息是分开存储的，每个 Topic 可以有多个生产者向它发送消息，也可以有多个消费者去消费其中的消息。
每个 Topic 可以划分多个分区（每个 Topic 至少有一个分区），同一 Topic 下的不同分区包含的消息是不同的。每个消息在被添加到分区时，都会被分配一个 offset，它是消息在此分区中的唯一编号，Kafka 通过 offset 保证消息在分区内的顺序，offset 的顺序不跨分区，即 Kafka 只保证在同一个分区内的消息是有序的。消息每次追加到对应的 Partition 的后面

安装

解压放到/opt/kafka, 软链一个latest出来, 先要启动zookeeper. 可以使用独立的zookeeper服务, 也可以用kafka自带的, 在lib目录下带了zookeeper. 启动zookeeper

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

然后启动kafka

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

此时数据日志在 /tmp/kafka-logs/ , 而应用日志的路径是 ./logs , 直接运行时, 会报 Cannot open file /opt/kafka/kafka_2.12-2.0.0/bin/../logs/kafkaServer-gc.log due to Permission denied 这样的错误.

修改数据日志路径

在 server.properties,  修改 log.dirs=/tmp/kafka-logs

修改应用日志路径

这个是在log4j.properties里指定的, 变量是 ${kafka.logs.dir},  可以通过设置环境变量 LOG_DIR 修改, 例如创建如下的sh脚本, 使用-daemon参数后, 启动后进入后台执行

export LOG_DIR=/tmp/kafka-application-logs/

echo 'Kafka application logs set to ' $LOG_DIR

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

在命令行测试Kafka

创建topic

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

读取topic列表

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

test

另外, 可以配置为自动创建topic, 当发布的topic不存在时自动创建

发布消息

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

This is a message

This is another message

启动消费端

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

This is a message

This is another message

应用场景

• 监控: 发送系统和应用程序健康相关的指标, 用独立的服务器收集和处理这些数据创建监控仪表盘并发送警告. LinkedIn还利用Apache Samza实现了一个能够实时处理事件的富调用图分析系统
• 传统的消息: 作为传统的消息队列实现消息的发布和订阅
• 分析: 为了更好地理解用户行为，改善用户体验，LinkedIn会将用户查看了哪个页面, 点击了哪些内容等信息发送到每个数据中心的Kafka集群上, 并通过Hadoop进行分析, 生成日常报告.
• 日志处理: 作为分布式应用程序或平台的构件, 大数据解决方案Pinot等产品将Kafka作为核心构件(分布式日志), 分布式数据库Espresso将其作为内部副本并改变传播层.

配置说明

ACKS

（1）acks=0: 设置为0表示producer不需要等待任何确认收到的信息。副本将立即加到socket buffer并认为已经发送。没有任何保障可以保证此种情况下server已经成功接收数据，同时重试配置不会发生作用（因为客户端不知道是否失败）回馈的offset会总是设置为-1；
（2）acks=1: 这意味着至少等待leader已经成功将数据写入本地log，但是不需等待所有follower是否成功写入。这种情况下，如果follower没有成功备份数据而leader又挂掉，则消息会丢失
（3）acks=all: 这意味着leader需要等待所有备份都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的保证
（4）其他的设置，例如acks=2也是可以的，这将需要给定的acks数量，但是这种策略一般很少用。

LINGER.MS

By default a buffer is available to send immediately even if there is additional unused space in the buffer. However if you want to reduce the number of requests you can set linger.ms to something greater than 0. This will instruct the producer to wait up to that number of milliseconds before sending a request in hope that more records will arrive to fill up the same batch.

producer组将会汇总任何在请求与发送之间到达的消息记录一个单独批量的请求。通常来说，这只有在记录产生速度大于发送速度的时候才能发生。然而，在某些条件下，客户端将希望降低请求的数量，甚至降低到中等负载一下。这项设置将通过增加小的延迟来完成--即，不是立即发送一条记录，producer将会等待给定的延迟时间以允许其他消息记录发送，这些消息记录可以批量处理。这可以认为是TCP种Nagle的算法类似。这项设置设定了批量处理的更高的延迟边界：如果消息达到了某个partition的batch.size，他将会立即发送而忽略这项设置，然而如果消息字节数比这项设置要小的多, 就会根据这个配置“linger”特定的时间以获取更多的消息。 这个设置默认为0，即没有延迟。设定linger.ms=5，例如，将会减少请求数目，但是同时会增加5ms的延迟。

BATCH.SIZE

This is an upper limit of how many messages Kafka Producer will attempt to batch before sending – specified in bytes.

producer将试图打包处理消息记录以减少请求次数。这将改善client与server之间的性能。这项配置控制默认的每批处理的消息字节数上限。不会试图打包大于这个字节数的消息字节数。
发送到brokers的请求将包含多个批量处理，其中会包含对每个partition的一个请求。
较小的批量处理数值比较少用，并且可能降低吞吐量(0则会仅用批量处理)。较大的批量处理数值将会浪费更多内存空间，这样就需要分配特定批量处理数值的内存大小。