ELK架构下利用Kafka Group实现Logstash的高可用

系统运维的过程中，每一个细节都值得我们关注

下图为我们的基本日志处理架构

所有日志由Rsyslog或者Filebeat收集，然后传输给Kafka，Logstash作为Consumer消费Kafka里边的数据，分别写入Elasticsearch和Hadoop，最后使用Kibana输出到web端供相关人员查看，或者是由Spark接手进入更深层次的分析。

在以上整个架构中，核心的几个组件Kafka、Elasticsearch、Hadoop天生支持高可用，唯独Logstash是不支持的，用单个Logstash去处理日志，不仅存在处理瓶颈更重要的是在整个系统中存在单点的问题，如果Logstash宕机则将会导致整个集群的不可用，后果可想而知

如何解决Logstash的单点问题呢？我们可以借助Kafka的Consumer Group来实现

Kafka Consumer Group

为了便于理解，我么先介绍一下Kafka里边几个重要的角色：

Broker：一台kafka服务器就是一个broker，一个kafka集群由多个broker组成，上图中的kafka集群有3台kafka服务器组成，也就是有3个broker，一个broker上可以有多个topic。

Topic：是个逻辑上的概念，用来区分不同的消息类别，类似于数据库中的表，可以将一组相同的数据发送给一个Topic，在日志处理中通常会将不同类型的日志写入不同的Topic，例如nginx日志写入名字为 nginx_log 的topic，tomcat日志写入名字为 tomcat_log 的topic，topic上图中没有标出，我们可以理解为图上的三个partition构成了一个topic

Partition：是kafka数据存储的基本物理单元，同一个Topic的数据可以被存储在一个或多个partition中，例如上图中的一个topic数据被存储在了partition1，partition2，partition3中，通常我们设置一个topic下partition的数量为broker的整数倍，这样一来数据能够均匀分布，二来可以同时利用集群下的所有服务器资源

Producer：生产者，向kafka写数据的服务，例如filebeat

Consumer：消费者，去kafka取数据的服务，例如logstash

Consumer Group：也是个逻辑上的概念，为一组consumer的集合，同一个topic的数据会广播给不同的group，同一个group中只有一个consumer能拿到这个数据

也就是说对于同一个topic， 每个group都可以拿到同样的所有数据，但是数据进入group后只能被其中的一个consumer消费 ，基于这一点我们只需要启动多个logstsh，并将这些logstash分配在同一个组里边就可以实现logstash的高可用了

input {
 kafka {
 bootstrap_servers => "10.8.9.2:9092,10.8.9.3:9092,10.8.9.4:9092"
 topics => ["ops_coffee_cn"]
 group_id => "groupA"
 codec => "json"
 }
}

以上为logstash消费kafka集群的配置，其中加入了 group_id 参数， group_id 是一个的字符串，唯一标识一个group，具有相同 group_id 的consumer构成了一个consumer group，这样启动多个logstash进程，只需要保证 group_id 一致就能达到logstash高可用的目的，一个logstash挂掉同一Group内的logstash可以继续消费

除了高可用外同一Group内的多个Logstash可以同时消费kafka内topic的数据，从而提高logstash的处理能力，但需要注意的是消费kafka数据时，每个consumer最多只能使用一个partition，当一个Group内consumer的数量大于partition的数量时， 只有等于partition个数的consumer能同时消费 ，其他的consumer处于等待状态。

例如一个topic下有3个partition，那么在一个有5个consumer的group中只有3个consumer在同时消费topic的数据，而另外两个consumer处于等待状态，所以想要增加logstash的消费性能，可以适当的增加topic的partition数量，但kafka中partition数量过多也会导致kafka集群故障恢复时间过长，消耗更多的文件句柄与客户端内存等问题，也并不是partition配置越多越好，需要在使用中找到一个平衡

kafka partition

kafka中partition数量可以在创建topic时指定：

# bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --create --topic ops_coffee --partitions 3
Created topic "ops_coffee".

--partitions：指定分区数，如果不指定默认会使用配置文件中 num.partitions 配置的数量

也可以手动修改partition的数量：

# bin/kafka-topics.sh --alter --zookeeper 127.0.0.1:2181 --partitions 5 --topic ops_coffee
Adding partitions succeeded!

注意partition的数量只能增加不能减少

如果想要知道topic的partition信息，可以通过以下命令查看topic详情：

# bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic ops_coffee
Topic:ops_coffee PartitionCount:3 ReplicationFactor:2 Configs:
 Topic: ops_coffee Partition: 0 Leader: 1 Replicas: 1,2 Isr: 1,2
 Topic: ops_coffee Partition: 1 Leader: 2 Replicas: 2,3 Isr: 2,3
 Topic: ops_coffee Partition: 2 Leader: 3 Replicas: 3,1 Isr: 3,1

至此对kafka consumer group有了更深入的了解，可以在具体的使用中游刃有余。

end：如果你觉得本文对你有帮助的话，记得关注点赞转发，你的支持就是我更新动力。