方案2：线上还有另一个传输方案，rsyslog设置一个中继，通过udp的方式将日志传到中继的rsyslog，由中继rsyslog入kafka，这么做的目的是方便了管理，当时还有个考虑是udp不会堵，但经过多轮测试后，nginx的rsyslog模块也是很健壮，不会堵的。
2、对于redis、php curl的日志
通过rsyslog的imfile模块，直接对文件监听，配置见上面的rsyslog配置，在日志轮转时通过超链接的方式进行新文件的连接，对应的超连接计划任务如下，每天0点5分执行：
5 0 * * * root sh /usr/local/script/php_slow_log.sh &> /dev/null
对应的php_slow_log.sh的脚本如下：
#!/bin/bash
DATE=`date +%F`
ln -sf /data1/ms/log/php_common/curl-$DATE /data1/ms/log/php_common/php_slow_log
备注：
a、rsyslog向4个kafka的broker推送消息时，是以轮训的方式；
b、rsyslog通过udp或tcp向外转发日志时，会默认加上时间、主机名、主机ip的属性。
二、队列（kafka+zookeeper）
队列用的是kafka，kafka集群使用zookeeper管理，我们用了4台服务器混装了4个kafka和3个zookeeper，kafka和zookeeper的安装地址如下：
注意：下载Binary downloads版本，别下错了，解压后就能用
1、关于kafaka
a、配置比较简单，基本默认即可,常调整的配置项如下：
配置文件：server.properties
broker.id=190 #id
num.partitions=20 #默认kafka的partion数量
log.dirs=/data1/kafka-logs #日志文件存放目录
log.retention.hours=3 #日志保留时间长短
zookeeper.connect=10.13.88.190:2181,10.13.88.191:2181,10.13.88.192:2181 #zookeeper指定
delete.topic.enable=true #topic是可以删除的
b、安装后测试(假设kafka和zookeeper都装了)：
开两个终端，两个终端分别运行如下命令
启动：./bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
终端1：./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
终端2：./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic test
注意两个终端的topic要一个名字，这时你在终端1输入任何数据，在终端2是同步的，证明你安装成功。
c、kafka常用管理命令
创建topic：./bin/kafka-topics.sh --create --topic test --replication-factor 1 --partitions 32 --zookeeper localhost:2181
删除topic：./bin/kafka-topics.sh --delete --topic test --zookeeper localhost:2181
查看topic列表：./bin/kafka-topics.sh --list --zookeeper localhost:2181
查看某个topic详细：./bin/kafka-topics.sh --describe --topic test --zookeeper localhosts:2181
监控某个topic的消费：./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test
指定消费组查看消费情况：./bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --group test
备注：topic下partitions的数量决定了并发消费的数量，在设置上要根据消息的QPS和硬盘情况合理配置。
2、关于zookeeper
a、配置比较简单，大多数默认项，最好奇数个，半数以上zookeeper存活可用
配置文件：zoo.cfg
dataDir=/data1/zookeeper
server.1=10.13.88.190:3888:4888
server.2=10.13.88.191:3889:4888
server.3=10.13.88.192:3889:4888
注意：要在数据目录手动建立myid，myid的值是server后面的数字，数字是有范围限制的1~255
b、zookeeper的常用管理命令
zookeeper我主要是看下它的整体状态，写了个简单脚本获取zookeeper的状态,执行结果如下：
脚本内容如下：
#!/bin/sh
#writer:gaolixu
[ -z $1 ] && echo "Please specify zoo.cfg like /usr/local/zookeeper/conf/zoo.cfg " && exit
cat $1 |grep "^server" |awk -F'[:|=]' '{print $2}' |
while read line
do
echo -ne "$line\t"
echo stat|nc -w 2 $line 2181 |egrep "^(Node|Zxid|Mode|Connections)" |tr "\n" "\t"
echo stat|nc -w 2 $line 2181 |egrep "^(Node|Zxid|Mode|Connections)" &>/dev/null || echo -n "host is done."
echo
done
使用方式：zkstat.sh /配置文件zoo.cfg的位置
zookeeper是相当稳定的，基本不用管。
备注：zookeeper配置文件里不能有汉字，否则启动不起来。
三、清洗（logstash）
logstash用做清洗，并且将处理好的日志推送到es里，安装过程很简单详见网址：
我线上的nginx的配置文件如下：
input {
kafka {
zk_connect => "10.13.88.190:2181,10.13.88.191:2181,10.13.88.192:2181"
topic_id => "cms-nginx"
group_id => "cms-nginx"
consumer_threads => 1
reset_beginning => false
decorate_events => false
}
}
filter {
ruby {