配置文件看着很长，其实阅读性很好，很易懂上手编写，无非就是定义切割点，如果大切割点下需要继续切割，就加if判断，继续切割，吐个槽里面threads和workers的数量好像不管用，我压测时去看线程数对不上,看的方法是top -H -p logstash的pid。
再就是看看哪些需要计算的变成数字型，还有个timestamp的处理，这个可以看看上面的代码，对于nginx打印的时间符合ISO8601标准，可以用他做es的时间索引，这样有个好处，如果某个环节慢索引赶不上的话，日志不会错序。时间标准详细可见：
备注：
a、尽量去掉没用的字段，精简索引，非常重要；
b、nginx打印出来的时间是标准化的，可以用它传到es作为timestamp建索引；
c、对于响应时间、响应内容大小、状态码要转换成数字类型，方便在kibana里做计算等操作；
d、切割双引号可以使用如下配置
code => "event.append(Hash[@kname.zip(event['message'].split(34.chr))])"
e、抓包后发现，logstash向es��数据是轮训的，从zookeeper消费数据不是轮训，可能是1个个用，有压力或问题后再去启用后面的zookeeper。
f、尽量按照官方如下写法建立多个索引向es推送，防止单个索引巨大，search时计算不出来
index => "logstash-cms-nginx-%{+YYYY.MM.dd.hh}"
g、测试性能方法如下
由于没有现成工具，我们用了打点计量的方式进行压测，摘掉es后将输出变为一个点，每处理一条信息打一个点，然后将打出的点用pv命令统计出字节流量，反推出logstash的吞吐量。
cp一个配置文件，修改output如下：
output {
stdout { codec => dots
workers => 1
}
}
同时为了不影响线上业务，修改group_id,这样的话测试消费和线上消费互不影响，配置文件修改如下：
kafka {
zk_connect => "10.13.88.190:2181,10.13.88.191:2181,10.13.88.192:2181"
topic_id => "nginx"
group_id => "test001"
consumer_threads => 12
reset_beginning => false
decorate_events => flase
}
测试时执行命令：/opt/logstash/bin/logstash -f /tmp/kafka_test.conf |pv -abt > /dev/null
压测结果如下：
每个点是一个byte，等到数据稳定后，计算每s的吞吐量为2.93*1024=3000，也就是这一个logstash最大吞吐量为能处理3000条信息每s。
四、索引（es 2.X 版本）
es是硬盘io和cpu消耗比较重的部分，硬优化有ssd，软优化牵涉到Java层面的GC调优、index调优、进程池调优、merge调优等，目前跑的还是比较好的，也有说将index和search分开的，防止search太大影响index索引，没去尝试，10多亿条日志，在目前的架构下性能还可以。
es的安装也是比较简单详见：https://www.elastic.co/guide/en/elasticsearch/reference/current/rpm.html
a、我们线上添加的进程调优如下，不加这个配置，跑一段时间就会有出现某一台es负载特别高的情况，而且难以恢复，甚至死机，加后运行健壮，先看调整前后对比图：