Kafka基础入门

1. Kafka简介

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消费。

2. 名词解释

• Broker
  Kafka集群包含一个或多个服务器，这种服务器被称为broker[5]
• Topic
  每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
• Partition
  Partition是物理上的概念，每个Topic包含一个或多个Partition.
• Producer
  消息生产者，负责发布消息到Kafka broker
• Consumer
  消息消费者，向Kafka broker读取消息的客户端。
• Consumer Group
  每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。

3. Kafka拓扑结构

一个典型的Kafka集群中包含若干Producer（可以是web前端产生的Page View，或者是服务器日志，系统CPU、Memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息。

对于传统的message queue而言，一般会删除已经被消费的消息，而Kafka集群会保留所有的消息，无论其被消费与否。当然，因为磁盘限制，不可能永久保留所有数据（实际上也没必要），因此Kafka提供两种策略删除旧数据。一是基于时间，二是基于Partition文件大小。例如可以通过配置$KAFKA_HOME/config/server.properties，让Kafka删除一周前的数据，也可在Partition文件超过1GB时删除旧数据，配置如下所示。

　　
# The minimum age of a log file to be eligible for deletion
log.retention.hours=168
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according to the retention policies
log.retention.check.interval.ms=300000
# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
log.cleaner.enable=false

4. Kafka常用命令

• 启动zookeeper

nohup /usr/kafka/kafka_2.11-0.11.0.0/bin/zookeeper-server-start.sh /usr/kafka/kafka_2.11-0.11.0.0/config/zookeeper.properties &

• 停止zookeeper

/usr/kafka/kafka_2.11-0.11.0.0/bin/zookeeper-server-stop.sh /usr/kafka/kafka_2.11-0.11.0.0/config/zookeeper.properties &

• 启动kafka

nohup /usr/kafka/kafka_2.11-0.11.0.0/bin/kafka-server-start.sh /usr/kafka/kafka_2.11-0.11.0.0/config/server.properties &

• 停止kafka

/usr/kafka/kafka_2.11-0.11.0.0/bin/kafka-server-stop.sh /usr/kafka/kafka_2.11-0.11.0.0/config/server.properties &

• 创建Topic

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

• 查看所有topic

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

• 查看某个topic主题详情

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test3

• 发送消息，使用生产者，创建一个broker

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

• 接收消息，使用消费者

./kafka-console-consumer.sh --zookeeper  localhost:2181 --topic test3 --from-beginning