豪翔天下

Change My World by Program

0%

Kafka 教程

重要概念

生产者(Producer)

消费者(Consumer)

消费消息。每个consumer属于一个特定的consumer group。使用consumer high level API时,同一个topic的一条消息只能被同一个consumer group内的一个consumer消费,但多个consumer group可同时消费这一消息。每个partition只会由一个consumer消费。

集群(Cluster)

宏观来看,Kafka主体包含的就是三部分: 生产者、消费者和集群,一个集群就是多个Broker的集合。

Broker

已经发布的消息就会保存在集群中的某个Broker中去。

Topic

用来区别message的种类,比如很多时候,与A相关的日志统一的topic定义为A,B相关的日志统一的topic定义为B,这样就不用一个一个单独地订阅了。物理上不通topic的消息分开存储,逻辑上一个topic的消息虽然保存于一个或多个broker上,但是用户只需指定消息的topic即可生产或消费数据而不必关心数据在哪里。

Partition

Kafka中每个Topic都会有一个或多个Partition,他是Kafaka数据存储的基本单元,每个Partition对应一个文件夹,文件夹下存储这个Partition的所有消息和索引。Kafka内部会根据算法得出一个值,根据这个值放入对应的partition目录中。所以读取时间复杂度为O(1)。分区的每一个消息都有一个连续的序列号叫做offset,用来在分区中唯一标识这个消息。一个topic可以保存在多个partition。Kafka会保证每个partition内部的顺序,但是不能保证跨partition的全局顺序,如果要保证全局有序,那么topic就只能有一个partition。如果一个group内部的consumer数量小于partition数量,那么至少有一个consumer会消费多个partition。当consumer数量和partition数量相等时效率最高。consumer数量不要大于partition数量,否则会有consumer空闲。consumer会自动负载到不同的partition。

  • 对于某一个topic,增加partition可以增加吞吐能力,但无法保证topic级别的有序性。

Segment

一个partition由多个Segment组成,一个Partition代表一个文件夹,一个Segment则代表该文件夹下的文件。Segment有大小限制,由log.segment.bytes 定义。

安装

docker方式安装

docker-compose.yml文件:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
version: '2'
services:
zookeeper:
image: wurstmeister/zookeeper
environment:
JMX: 9000
ports:
- "2181:2181"
kafka:
image: wurstmeister/kafka # 这个镜像使用文档见https://github.com/wurstmeister/kafka-docker
ports:
- "9092"
environment:
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://:9092 # 这是重点,否则,在容器内部启动生产者消费者都会失败的
KAFKA_LISTENERS: PLAINTEXT://:9092
KAFKA_CREATE_TOPICS: "test:1:1" # 自动创建一个默认的topic
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" # 禁用掉自动创建topic的功能,使用上面的镜像,kafka的参数设置都可以以这样的方式进行设置
volumes:
- /var/run/docker.sock:/var/run/docker.sock
kafka-manager:
image: sheepkiller/kafka-manager # 如果要安装web管理工具可以同时安装这个,最后通过宿主机IP的9000端口进行访问,例如172.31.148.174:9000
links:
- kafka
- zookeeper
environment:
ZK_HOSTS: zookeeper:2181
APPLICATION_SECRET: "letmein"
ports:
- "9000:9000"
expose:
- "9000"

安装命令:

1
2
3
4
5
docker-compose up -d			# 默认只会有一个kafka实例
docker-compose scale kafka=n # 将kafka实例增加到n个,什么都不用修改,就能直接建立一个集群
docker-compose stop # 暂停所有容器
docker-compose start # 开启所有容器
docker-compose rm -f # 删除所有容器

kafka命令

kafka-console-consumer.sh

1
kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic test	# 启动一个消费者,监听test这个topic

kafka-console-producer.sh

1
2
kafka-console-producer.sh --broker-list localhost:9092 --topic test	# 启动一个生产者,直接输入消息回车即可发送消息了
kafka-console-producer.sh --broker-list localhost:9092 --topic test < access.log # 直接将文件内容传入kafka

kafka-consumer-groups.sh

1
2
kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --list	# 查看新消费者列表
kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --describe --group kafka-python-default-group # 查看某消费者的消费详情,这里的消费者名称就是kafka-python-default-group

kafka-producer-perf-test.sh自带的压测工具

1
kafka-producer-perf-test.sh --topic test --num-records 10000 --record-size 1 --throughput 100  --producer-props bootstrap.servers=localhost:9092	# 总共100条数据,每条大小是1

kafka-topics.sh

1
2
kafka-topics.sh --list --zookeeper zookeeper:2181		# 列出所有的topic
kafka-topics.sh --describe --zookeeper zookeeper:2181 # 查看集群描述

安全认证

Kafka可以配合SSL+ACL来进行安全认证: http://orchome.com/185

TroubleShooting

  • 容器内部启动生产者出现错误:[2016-12-26 03:03:39,983] WARN Error while fetching metadata with correlation id 0 : {test=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient)

    是因为docker-compose文件里面的宿主讥IP设置出错,如果是动态IP的话就没办法了,只能删除重新创建了

  • 启动生产者或者消费者出现LEADER_NOT_AVAILABLE:原因是没有执行docker-compose scale kafka=n

拓展阅读
坚持原创技术分享,谢谢支持

欢迎关注我的其它发布渠道