kafka入门

一种分布式的、基于发布/订阅的消息系统，scala编写，具备快速、可扩展、可持久化的特点。

基本概念

topic
主题

partition
分区，一个topic下可以有多个partition，消息是分散到多个partition里存储的，partition支持水平扩展。
一个partition内的消息是有序的，partition间的消息则是无序的。
每个partition会有若干副本。

broker
一个kafka节点

consumer
消费者，从topic里取得消息。
每个consumer维护自己的offset。
consumer数量小于分区数，会有一个消费者处理多个分区；反之，会有空闲的消费者，造成浪费。

producer
生产者，负责将消息写入topic

特点

基于硬盘的消息保存，避免在producer上累积消息或者消息丢失。
同一个消息可以由多个consumer消费。

可扩展性：随着数据的增加，可扩展为数十台，上百台规模的大集群。扩展可以在集群正常运行的时候进行，对于整个系统的运作没有影响；这也就意味着，对于很多台broker 的集群，如果一台broker 有故障，不影响为client 提供服务.集群如果要同时容忍更多的故障的话, 可以配置更高的replication
factors。

高性能：上面的这些特性使得Apache Kafka 成为一个能够在高负载的情况下表现出优越性能的发布-订阅消息系统。Producer, consumer 和broker 都能在大数据流的情况下轻松的扩展.

kafka的版本

比如kafka_2.10-0.10.2.0
这里，2.10指的是编译kafka的scala版本，真正的kafka版本号是后面的：0.10.2.0

配置kafka

config/server.properties文件里的配置项说明：
broker.id
每个kafka 的broker 都需要有一个整型的唯一标识，这个标识通过broker.id 来设置。默认的情况下，这个数字是0，但是它可以设置成任何值。需要注意的
是，需要保证集群中这个id 是唯一的。这个值是可以任意填写的，并且可以在必要的时候从broker 集群中删除。比较好的做法是使用主机名相关的标识来做
为id，比如，你的主机名当中有数字相关的信息，如hosts1.example.com，host2.example.com，那么这个数字就可以用来作为broker.id 的值。

port
默认启动kafka 时，监听的是TCP 的9092 端口，端口号可以被任意修改。如果端口号设置为小于1024，那么kafka 需要以root 身份启动。但是并不推荐以root 身份启动。

zookeeper.connect
这个参数指定了Zookeeper 所在的地址，它存储了broker 的元信息。默认是运行在本机的2181 端口上，因此这个值被设置成
localhost:2181。这个值可以通过分号设置多个值，每个值的格式都是hostname:port/path

log.dirs
这个参数用于配置Kafka 保存数据的位置，Kafka 中所有的消息都会存在这个目录下。可以通过逗号来指定多个目录，kafka 会根据最少被使用的原则选择目录分配新的partition。注意kafka 在分配partition 的时候选择的规则不是按照磁盘的空间大小来定的，而是分配的parition 的个数大小。

num.recovery.thread.per.data.dir
kafka 可以配置一个线程池，线程池的使用场景如下：

• 当正常启动的时候，开启每个parition 的文档块
• 当失败后重启时，检查parition 的文档块
• 当关闭kafka 的时候，清除关闭文档块
  默认，每个目录只有一个线程。最好是设置多个线程数，这样在服务器启动或者关闭的时候，都可以并行的进行操作。尤其是当非正常停机后，重启时，如果有大量的分区数，那么启动broker 将会花费大量的时间。
  【注意】这个参数是针对每个目录的。比如，num.recovery.threads.per.data.dir 设置为8，如果有3个log.dirs 路径，那么一共会有24 个线程。

num.partitions
这个参数用于配置新创建的topic 有多少个分区，默认是1 个。注意partition 的个数只可以被增加，不能被减少。这就意味着如果想要减少主题的分区数，那
么就需要重新创建topic。
在第一章中介绍过，kafka 通过分区来对topic 进行扩展，因此需要使用分区的个数来做负载均衡，如果新增了broker，那么就会引发重新负载分配。这并不意味着所有的主题的分区数都需要大于broker 的数量，因为kafka 是支持多个主题的，其他的主题会使用其余的broker。需要注意的是，如果消息的吞吐量很高，那么可以通过设置一个比较大的分区数，来分摊压力。

log.retention.ms
这个参数用于配置kafka 中消息保存的时间，也可以使用log.retention.hours，默认这个参数是168 个小时，即一周。另外，还支持log.retention.minutes 和log.retention.ms。这三个参数都会控制删除过期数据的时间，推荐还是使用log.retention.ms。如果多个同时设置，那么会选择最小的那个。

log.retention.bytes
这个参数也是用来配置消息过期的，它会应用到每个分区，比如，你有一个主题，有8 个分区，并且设置了log.retention.bytes 为1G，那么这个主题总共可以保留8G 的数据。注意，所有的过期配置都会应用到patition 粒度，而不是主题粒度。这也意味着，如果增加了主题的分区数，那么主题所能保留的数据也就随之增加了。
如果设置了log.retention.bytes 和log.retention.ms（或者其他过期时间的配置），只要满足其中一个条件，消息就会被删除。

log.segment.bytes
这个参数用来控制log 段文件的大小，而不是消息的大小。在kafka 中，所有的消息都会进入broker，然后以追加的方式追加到分区当前最新的segment 段文件中。一旦这个段文件到达log.segment.bytes 设置的大小，比如默认的1G，这个段文件就会被关闭，然后创建一个新的。一旦这个文件被关闭，就可以理解成这个文件已经过期了。这个参数设置的越小，那么关闭文件创建文件的操作就会越频繁，这样也会造成大量的磁盘读写的开销。
通过生产者发送过来的消息的情况可以判断这个值的大小。比如，主题每天接收100M 的消息，并且log.segment.bytes 为默认设置，那么10 天后，这个段文件才会被填满。由于段文件在没有关闭的时候，是不能删除的，log.retention.ms 又是默认的设置，那么这个消息将会在17 天后，才过期删除。因为10 天后，段文件才关闭。再过7 天，这个文件才算真正过期，才能被清除。

message.max.bytes
这个参数用于限制生产者消息的大小，默认是1000000，也就是1M。生产者在发送消息给broker 的时候，如果出错，会尝试重发；但是如果是因为大小的原因，那生产者是不会重发的。另外，broker上的消息可以进行压缩，这个参数是指压缩后的大小，这样能多存储很多消息。
需要注意的是，允许发送更大的消息会对性能有很大影响。更大的消息，就意味着broker 在处理网络连接的时候需要更长的时间，它也会增加磁盘的写操作压力，影响IO 吞吐量。

启动kafka

进入kafka/bin目录。

启动zk

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

测试zk是否启动

telnet localhost 2181
输入srvr，应该会返回：

[2017-12-01 15:59:18,829] INFO Processing srvr command from /0:0:0:0:0:0:0:1:40194 (org.apache.zookeeper.server.NIOServerCnxn)
Zookeeper version: 3.4.6-1569965, built on 02/20/2014 09:09 GMT
Latency min/avg/max: 0/0/0
Received: 1
Sent: 0
Connections: 1
Outstanding: 0
Zxid: 0xdf
Mode: standalone
Node count: 127
[2017-12-01 15:59:18,833] INFO Closed socket connection for client /0:0:0:0:0:0:0:1:40194 (no session established for client) (org.apache.zookeeper.server.NIOServerCnxn)
Connection closed by foreign host.

不建议zookeeper运行在多于7个的节点上，因为集群性能会因一致性的特征而降低。

启动kafka

nohup  ./kafka-server-start.sh -daemon ../config/server.properties

以守护进程方式执行。

topic管理命令

创建具有指定数量分区（或复制因子）的topic
./kafka-topics.sh --create --zookeeper xx.xx.xx.xx:2181 --topic test1 --replication-factor 1 --partitions 6

查看topic的元信息
./kafka-topics.sh --zookeeper localhost:2181 --describe --topic test1

查看所有topic
./kafka-topics.sh --zookeeper localhost:2181 --list

修改已有topic的分区

./kafka-topics.sh --alter --zookeeper localhost:2181 --topic test1 --partitions 4./kafka-topics.sh --zookeeper xx.xx.xx.xx:2181 --alter --topic test1 --config flush.ms=5000./kafka-topics.sh --zookeeper xx.xx.xx.xx:2181 --alter --topic test1 --config flush.messages=20000

删除topic

./kafka-topics.sh --delete --zookeeper xx.xx.xx.xx:2181 --topic test1

但删除topic并不是真的删除，只是打个标记，要真正删除，需要同时修改server.properties：

delete.topic.enable=true

并重启kafka才能生效。

向topic发布消息

./kafka-console-producer.sh --broker-list localhost:9092 --topic test1
输入：
msg1
msg2
按^D停止发送。

若kafka的server.properties配置了host.name，则localhost必须改成host.name的值，例如：
./kafka-console-producer.sh --broker-list xx.xx.xx.xx:9092 --topic test1

查看topic里的消息

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test1 --from-beginning
输出：
msg1
msg2
按^C停止接收。