kafka原理及Docker环境部署
技术原理
Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka为处理实时数据提供一个统一、高吞吐、低延迟的平台。其持久化层本质上是一个“按照分布式事务日志架构的大规模发布/订阅消息队列”，这使它作为企业级基础设施来处理流式数据非常有价值。此外，Kafka可以通过Kafka Connect连接到外部系统（用于数据输入/输出），并提供了Kafka Streams——一个Java流式处理库 (计算机)。
Kafka是一个分布式的、高吞吐量、高可扩展性的消息系统。Kafka 基于发布/订阅模式，通过消息解耦，使生产者和消费者异步交互，无需彼此等待。Ckafka 具有数据压缩、同时支持离线和实时数据处理等优点，适用于日志压缩收集、监控数据聚合等场景。
关键名词：
-
broker：kafka集群包含一个或者多个服务器，服务器就称作broker
-
producer：负责发布消息到broker
-
consumer：消费者，从broker获取消息
-
topic：发布到kafka集群的消息类别。
-
partition：每个topic划分为多个partition。
-
group：每个partition分为多个group
架构示意图
image
一个典型的Kafka集群中包含若干Producer（可以是web前端FET，或者是服务器日志等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干ConsumerGroup，以及一个Zookeeper集群。
Kafka通过Zookeeper管理Kafka集群配置：选举Kafka broker的leader，以及在Consumer Group发生变化时进行rebalance，因为consumer消费kafka topic的partition的offsite信息是存在Zookeeper的。
Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息。
image
一个典型的Cloud Kafka集群如上所示。其中的生产者Producer可能是网页活动产生的消息、或是服务日志等信息。生产者通过push模式将消息发布到Cloud Kafka的Broker集群，消费者通过pull模式从broker中消费消息。消费者Consumer被划分为若干个Consumer Group，此外，集群通过Zookeeper管理集群配置，进行leader选举，故障容错等。
kafka特点：
- 它是一个处理流式数据的”发布-订阅“消息系统。
- 实时高效处理流式数据：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。
- 将数据安全存储在分布式集群。
- 它是运行在集群上的。
- 它将流式记录存储在topics中。
- 每个record由key, value和timestamp组成。
Docker搭建
参考：https://github.com/wurstmeister/kafka-docker
docker-compose.yml如下：
version: '2'
services:
zookeeper:
image: wurstmeister/zookeeper
volumes:
- ./data:/data
ports:
- "2181:2181"
kafka:
image: wurstmeister/kafka
ports:
- "9092:9092"
environment:
KAFKA_ADVERTISED_HOST_NAME: 10.154.38.115
KAFKA_MESSAGE_MAX_BYTES: 2000000
KAFKA_CREATE_TOPICS: "Topic1:1:3,Topic2:1:1:compact"
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
volumes:
- ./kafka-logs:/kafka
- /var/run/docker.sock:/var/run/docker.sock
kafka-manager:
image: sheepkiller/kafka-manager
ports:
- 9020:9000
environment:
ZK_HOSTS: zookeeper:2181
参数说明：
- KAFKA_ADVERTISED_HOST_NAME：Docker宿主机IP（如果你要配置多个brokers，就不能设置为 localhost 或 127.0.0.1）
- KAFKA_MESSAGE_MAX_BYTES：kafka(message.max.bytes) 会接收单个消息size的最大限制，默认值为1000000 , ≈1M
- KAFKA_CREATE_TOPICS：初始创建的topics，可以不设置
- 环境变量./kafka-logs为防止容器销毁时消息数据丢失。
- 容器kafka-manager为yahoo出可视化kafka WEB管理平台。
操作命令：
# 启动：
$ docker-compose up -d
# 增加更多Broker：
$ docker-compose scale kafka=3
# 合并：
$ docker-compose up --scale kafka=3
Kakfa使用
1，Kafka管理节点
image
2，主题
environment:
KAFKA_CREATE_TOPICS: "Topic1:1:3,Topic2:1:1:compact"
Topic1有1个Partition和3个replicas， Topic2有2个Partition，1个replica和cleanup.policy为compact。
Topic 1 will have 1 partition and 3 replicas, Topic 2 will have 1 partition, 1 replica and a cleanup.policy set to compact.
3，读写验证
读写验证的方法有很多，这里我们用kafka容器自带的工具来验证，首先进入到kafka容器的交互模式：
docker exec -it kafka_kafka_1 /bin/bash
创建一个主题：
/opt/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.31.84:2181 --replication-factor 1 --partitions 1 --topic my-test
查看刚创建的主题：
/opt/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.31.84:2181
image
发送消息：
/opt/kafka/bin/kafka-console-producer.sh --broker-list 192.168.31.84:9092 --topic my-test
This is a message
This is another message
读取消息：
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.31.84:9092 --topic my-test --from-beginning
使用场景
- 日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。
image
- 消息系统：解耦和生产者和消费者、缓存消息等。
- 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
- 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
- 流式处理：比如spark streaming和storm
参考：
1，https://www.jianshu.com/p/bfeceb3548ad
2，https://www.jianshu.com/p/7f089cdff29a
3，https://www.cnblogs.com/iforever/p/9130983.html
4，利用flume+kafka+storm+mysql构建大数据实时系统
5，Kafka系列（四）Kafka消费者：从Kafka中读取数据
6，基于Docker搭建分布式消息队列Kafka
