背景: 直接使用EFK进行日志收集，在大规模高压力的情况下Elasticsearch会存在丢数据的情况，现在考虑使用MQ（Message Queue）进行缓冲，达到不丢数据的目的。由于对于日志收集响应速度并不是十分高，并且对日志的可靠性要求较高，最终选择Kafka来充当消息队列而非官方推荐的redis。这里着重进行kafka介绍，之后会整合EFK+kafka的应用落地记录。

关于Kafka的基本原理

基本介绍

Kafka是由LinkedIn使用Scala开发的一个分布式的消息系统。最初用作LinkedIn的活动流（Activity Stream）和运营数据处理管道（Pipeline），Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。kafka是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统(也可以当做MQ系统)，常见可以用于web/nginx日志、访问日志，消息服务等等。

基本组成

Tips： 主要由四部分组成，Topic（话题）、Producer（生产者）、Broker（服务节点）、Consumer（消费者）

  • Broker

    已发布的消息保存在一组服务器中，它们被称为代理（Broker）或Kafka集群。组成kafka集群的每个服务器，都称为是Broker。Broker可以容纳多个Topic

  • Topic

    是特定类型的消息流。消息是字节的有效负载（Payload），话题是消息的分类名或种子（Feed）名。每条发送到kafka的消息都有一个类别，这个类别就叫做Topic。可以理解为一个消息队列（Message Queue）的名称。（物理上不同 Topic 的消息分开存储，逻辑上一个 Topic 的消息虽然保存于一个或多个 broker 上，但用户只需指定消息的 Topic 即可生产或消费数据而不必关心数据存于何处）。 Partition

    • parition是物理上的概念，每个topic包含一个或多个partition，创建topic时可指定parition数量。每个partition对应于一个文件夹，该文件夹下存储该partition的数据和索引文件
    • partition分区数，控制topic将分片成多少个log。可以显示指定，如果不指定则会使用broker(server.properties)中的num.partitions配置的数量
    • 为了实现扩展性，一个非常大的topic可以分布到多个 broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。
    • partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体 （多个partition间）的顺序。
    • 也就是说，一个topic在集群中可以有多个partition，那么分区的策略是什么？(消息发送到哪个分区上，有两种基本的策略，一是采用Key Hash算法，一是采用Round Robin算法)

    Offset

    • kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka
  • Producer

    消息生产者，就是向kafka broker发消息的客户端。负责发布消息到Kafka broker。

  • Consumer

    消息消费者，向 Kafka broker 读取消息的客户端。每个consumer属于一个特定的consuer group（可为每个consumer指定group name，若不指定group name则属于默认的group）。使用consumer high level API时，同一topic的一条消息只能被同一个consumer group内的一个consumer消费，但多个consumer group可同时消费这一消息。 Consumer Group

    • 每个 Consumer 属于一个特定的 Consumer Group（可为每个 Consumer 指定 group name，若不指定 group name 则属于默认的 group）
    • 消息系统有两类，一是广播，二是订阅发布。广播是把消息发送给所有的消费者；发布订阅是把消息只发送给订阅者。Kafka通过Consumer Group组合实现了这两种机制： 实现一个topic消息广播（发给所有的consumer）和单播（发给任意一个consumer）。一个topic可以有多个Consumer Group。
    • topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个CG只会把消息发给该CG中的一个 consumer（这是实现一个Topic多Consumer的关键点：为一个Topic定义一个CG，CG下定义多个Consumer）。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。
    • 典型的应用场景是，多个Consumer来读取一个Topic(理想情况下是一个Consumer读取Topic的一个Partition）,那么可以让这些Consumer属于同一个Consumer Group即可实现消息的多Consumer并行处理，原理是Kafka将一个消息发布出去后，ConsumerGroup中的Consumers可以通过Round Robin的方式进行消费(Consumers之间的负载均衡使用Zookeeper来实现)

搭建kafka集群

**Tips：**这里使用的zk及kafka版本如下 zookeeper:3.4.10 kafka:2.11-0.10.2.0

安装配置zookeeper

  • Install 
    1
2
3
4
5
6
7
8

    wget http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz
tar xvf zookeeper-3.4.10.tar.gz
cp zookeeper-3.4.10 /usr/local/zookeeper/ -r
cd /usr/local/zookeeper/
cp conf/zoo_sample.cfg conf/zoo.cfg
mkdir -p /var/zookeeper/data
echo 1 >/var/zookeeper/data/myid
#注意，这里要与下面config中server.1的配置一致，此server为10.17.0.112，所以echo 1，同理226这台server就echo 2
  • Config 
    1
2
3
4
5
6
7
8
9

    #cat /usr/local/zookeeper/conf/zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/var/zookeeper/data
clientPort=2181
server.1=10.17.0.112:2888:3888
server.2=10.17.0.226:2888:3888
server.3=10.17.0.211:2888:3888
  • Start 
    1
2
3

    cd /usr/local/zookeeper
./bin/zkServer.sh start
./bin/zkServer.sh status

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

    #cat /etc/rc.d/init.d/zookeeper
#!/bin/bash
#chkconfig:2345 20 90
#description:zookeeper
#processname:zookeeper
case $1 in
      start) /usr/local/zookeeper/bin/zkServer.sh start;;
      stop) /usr/local/zookeeper/bin/zkServer.sh stop;;
      status) /usr/local/zookeeper/bin/zkServer.sh status;;
      restart) /usr/local/zookeeper/bin/zkServer.sh restart;;
      *)  echo "require start|stop|status|restart";;
esac

安装配置kafka

  • Install 
    1
2
3
4

    wget http://mirror.bit.edu.cn/apache/kafka/0.10.2.0/kafka_2.11-0.10.2.0.tgz
tar xvf kafka_2.11-0.10.2.0.tgz
cp kafka_2.11-0.10.2.0 /usr/local/kafka/ -r
cd /usr/local/kafka/
  • Config（需要改server和consumer配置） 
     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

    #cat /usr/local/kafka/config/server.properties |egrep -v "^#|^$"
broker.id=2
#唯一值，我这里按zk的serverID进行了配置
listeners = PLAINTEXT://10.17.0.226:9092
#这个地方在这个版本及以后需要打开注释填写本机地址，之前的版本需配置host.name
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.17.0.112:2181,10.17.0.226:2181,10.17.0.221:2181
#这里填写zookeeper的地址，另外消费配置上也需要做对应修改
zookeeper.connection.timeout.ms=6000

    1
2
3
4

    #cat /usr/local/kafka/config/consumer.properties |egrep -v "^#|^$"
zookeeper.connect=10.17.0.112:2181,10.17.0.226:2181,10.17.0.221:2181
zookeeper.connection.timeout.ms=6000
group.id=test-consumer-group
  • Start 
    1
2
3
4
5

    cd /usr/local/kafka/
bin/kafka-server-start.sh config/server.properties &
#后台执行，退出终端后终止
bin/kafka-server-start.sh -daemon config/server.properties &
#后台执行，退出终端后不终止

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

    #cat /etc/rc.d/init.d/kafka
#!/bin/bash
#chkconfig:2345 30 80
#description:kafka
#processname:kafka
case $1 in
 start) /usr/local/kafka/bin/kafka-server-start.sh  -daemon /usr/local/kafka/config/server.properties &;;
 stop) /usr/local/kafka/bin/kafka-server-stop.sh  /usr/local/kafka/config/server.properties;;
 restart) /usr/local/kafka/bin/kafka-server-stop.sh  /usr/local/kafka/config/server.properties
         /usr/local/kafka/bin/kafka-server-start.sh  -daemon /usr/local/kafka/config/server.properties &
         ;;

 *)  echo "require start|stop|restart"  ;;
esac

安装配置kafka监控程序

目前常用有三种： Kafka Web Console：监控功能较为全面，可以预览消息，监控Offset、Lag等信息，但存在bug，不建议在生产环境中使用。 Kafka Manager：偏向Kafka集群管理，若操作不当，容易导致集群出现故障。对Kafka实时生产和消费消息是通过JMX实现的。没有记录Offset、Lag等信息。 KafkaOffsetMonitor：程序一个jar包的形式运行，部署较为方便。只有监控功能，使用起来也较为安全。 这里使用第三种KafkaOffsetMonitor

  • install 
    1
2

    wget https://github.com/quantifind/KafkaOffsetMonitor/releases/download/v0.2.1/KafkaOffsetMonitor-assembly-0.2.1.jar
#下载即可
  • start 
    1
2
3
4
5
6

    java -cp KafkaOffsetMonitor-assembly-0.2.1.jar \
     com.quantifind.kafka.offsetapp.OffsetGetterWeb \
     --zk 10.17.0.112:2181,10.17.0.226:2181,10.17.0.221:2181 \
     --port 8089 \
     --refresh 10.seconds \
     --retain 2.days &

常用kafka命令

  • 创建及查看Topic 
    1
2
3
4
5
6

    bin/kafka-topics.sh --create --zookeeper 10.17.0.211:2181 --replication-factor 3 --partitions 2 --topic prod-test
#创建一个名为prod-test的topic，有3个副本（即控制消息保存在3个broker上），2个分区
bin/kafka-topics.sh --list --zookeeper 10.17.0.211:2181
#list topic，列出所有topic
bin/kafka-topics.sh --describe --zookeeper 10.17.0.211:2181 --topic prod-test
#查看某个topic的具体信息
  • 生产消费数据（可以测试集群搭建是否成功） 
    1
2
3
4

    bin/kafka-console-producer.sh --broker-list 10.17.0.211:9092 --topic prod-test
#从控制台向topic生产数据
bin/kafka-console-consumer.sh  --zookeeper 10.17.0.226:2181  --topic prod-test --from-beginning
#从控制台消费topic prod-test的数据，可以使用任意集群中的地址进行测试
  • 查看topic某分区偏移量最大（小）值 
    1
2
3
4

    bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic prod-test  --time -2 --broker-list 10.17.0.211:9092 --partitions 1
#查看prod-test这个topic分区1的分组最小偏移值（time -2 表示最小偏移量）
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic prod-test  --time -1 --broker-list 10.17.0.211:9092 --partitions 1
#查看prod-test这个topic分区0的分组最大偏移值（time -1 表示最大偏移量
  • 增加topic分区数 
    1
2

    bin/kafka-topics.sh --zookeeper 10.17.0.211:2181  --alter --topic prod-test --partitions 5
#增加prod-test的这个topic的分区数到5个（比如之前是3个，就是再增加2个），这个数字只能比现在已有的分区数大。
  • 查看topic消费进度 
    1
2

    #consumer group可以从zk中查看
bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group console-consumer-50561  --zookeeper 10.17.0.226:2181