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

日志处理架构

所有日志由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

为了便于理解，我们先介绍一下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 => "1.1.1.1:9092,2.2.2.2:9092,3.3.3.3:9092"
        topics => ["Mysia"]
        group_id => "group1"
        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 Mysia --partitions 4
Created topic "Mysia".

--partitions： 指定分区数，如果不指定默认会使用配置文件中num.partitions配置的数量
也可以手动修改partition的数量：

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

注意partition的数量只能增加不能减少，如果想要知道topic的partition信息，可以通过以下命令查看topic详情：

# bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic Mysia
Topic:Mysia 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

完。