一、Logstash简介
Logstash是一个开源数据收集引擎，具有实时管道功能。Logstash可以动态地将来自不同数据源的数据统一起来，并将数据标准化到你所选择的目的地。
二、Logstash处理流程
Logstash处理流程大致可分为3个阶段，Input---->Filter---->Output,用中文描述一下分别是，数据采集----->数据分析/解析---->数据输出；具体的处理流程可以查看下图，下面的一些函数和一些概念等后面我们在具体讲讲：
这里我们解释一下codec的概念，Codec 是 logstash 从 1.3.0 版开始新引入的概念(Codec 来自 Coder/decoder 两个单词的首字母缩写)。在此之前，logstash 只支持纯文本形式输入，然后以过滤器处理它。但现在，我们可以在输入 期处理不同类型的数据，这全是因为有了 codec 设置。所以，这里需要纠正之前的一个概念。Logstash 不只是一个input | filter | output 的数据流，而是一个 input | decode | filter | encode | output 的数据流！codec 就是用来 decode、encode 事件的。
Logstash Event也需要介绍，类似Java的Object的对象，在这个过程我们可以对其进行一些赋值等操作；
三、Logstash架构介绍
上面介绍一下Logstash的数据流向，接下来我们介绍下Logstash的架构，当然我介绍的是6.X的架构，看下图
由上图得知，我们可以有多个输入的文本，另外由Queue分发到不同的Pipline中，这里的Pipline中文的意思就是管道，在程序中我们可以理解为线程，可以有多个Pipline，并且每个Pipline之间是互不打扰的，另外每个Batcher，Filter和Output组成，Batcher是批量从Queue获取数据的，这个值可以通过配置进行设置；
Pipline配置如下:
pipeline.workers : 8 Pipeline线程数，及filter_output的处理线程数，默认是CPU的核数，命令行参数为-w
pipeline.batch.size : 125 Batcher一次批量获取的待处理文档数，默认125（建议向es输出的时候10-20Mb之间，可以计算一下文档数），可以根据输出进行调整，越大会占用越多的heap空间，可以通过jvm.options调整。命令行参数-b
pipeline.batch.delay :5 ：Batcher的等待时长，单位为ms。命令行参数-u
接下来我们在了解下Queue设计思路，使用Queue以后我们首先要关注的是怎么确保数据是正常被消费掉了，这里到达QutPut以后会有发送一个ACK给Queue来告诉Queue这些Logstash Event已经处理完了，这样就能确保一个数据正常被消费掉，这也是在消息队列中我们常用的一种手段，接下来我们再聊聊Queue分类:
1.In Memory（在内存）
在内存中，固定大小，无法处理进程crash、机器宕机等情况，会导致数据丢失。
2.Persistent Queue In Disk (持久化到磁盘)
可处理进程crash情况，保证数据不丢失。保证数据至少消费一次；充当缓冲区，可代替kafka等消息队列作用。
接下来我们来了解下持久队列是怎么保证的?这里我们从数据到队列以后的处理开始说起，首先队列将数据备份到disk(磁盘)，队列返回响应给Input,最后数据经过OutPut以后返回ACK给队列，当队列接收到消息以后开始删除disk中备份的数据，这样就能保证数据持久性；
性能方面对比如下图，基本性能是无差别的：
队列主要配置如下：
queue.type: persisted 默认是memory
queue.max_bytes: 4gb 默认是1gb
四、Logstash配置介绍
我们会使用的配置文件在config目录录下面，logstash.yml和官方，接下来我们主要介绍下参数的配置logstash.yml配置：jvm.options,另外6.0以上的版本会有pipelines.yml,这个文件是为了在同一个进程中运行多个管道，具体可以参考下
node.name：节点名，默认是主机名；
path.data:持续化存储数据的文件夹，默认是在Logstash home目录下的data
path.config:设定Pipeline配置文件的目录
path.log:设置Pipeline日志目录
第三小节介绍的Queue和Pipeline都可以在该文件下设置，这里不做过多的介绍，另外更细节的一些参数配置大家可以参考官方；
jvm.options这个里面的配置大家可以自己根据自己机器的情况进行下JVM参数调优；
另外还有Pipeline配置文件，定义数据处理流程的文件，以.conf结尾
五、Pipeline配置
布尔类型 Boolean：isFailed => true
数值类型 Number：age => 33
字符串类型 String: name => “Hello”
数组类型：users => [{age=>11, name=>wtz}, {age => 22, name => myt}] 或者 path => [“/var/log/error,log”,”/var/log/warn.log”]
哈希类型：match => {“field” => “value1” “field”=>”value2”}
备注：#
在配置中可以引用Logstash Event的属性（字段）,主要由下面两种方式：
1.直接引用字段值 – 直接使用[],如果是多层直接嵌套
2.在字符串中以 sprintf 方式引用 – 使用%{}来实现
配置文件支持条件判断语法：
表达式操作如下：
1.比较：== != < > <= >=
2.正则是否匹配：=~ =
3.包含（字符串或者数组）：in, not in
4.布尔操作符：and or nand xor !
5.分组操作符(条件非常复杂的时候，其实就相当于括号)：()
六、插件详解
第二部分的时候我们介绍了Logstash的数据处理流程，涉及到一些插件流程，接下来我们介绍下插件的配置，也是分成3部分介绍:
input插件介绍:
1.stdin
最简单的输入，从标准输入读取数据
2.file
从文件读取数据，参数介绍如下:
3.kafka
这个是我们使用的配置，供大家参考下，另外大家可以参考官方文档
input { kafka { bootstrap_servers => "服务地址" group_id => "消费组id" topics => ["订阅的主题列表","订阅的主题列表"] codec => "json" auto_offset_reset => "Kafka偏移量，超出设置为最早偏移量" consumer_threads => "线程数" } }
另外大家还可以看下官方文档选择自己合适的使用；
Filter插件介绍
1.grok
解析和构造任意文本，Grok是目前Logstash中解析非结构化日志数据到结构化和可查询数据的最佳方式，使用内置的120种模式；
另外可以阅读下这篇文章你真的理解grok吗
2.mutate
对事件字段执行一般的转换，你可以重命名、删除、替换和修改事件中的字段。
3.drop
完全删除事件，例如调试事件。
4.clone
复制事件，可能添加或删除字段。
5.geoip
添加关于IP地址地理位置的信息
更多的大家可以阅读官方文档
OutPut插件介绍
最常用的就是输出到es,配置如下
elasticsearch { hosts => "地址" index => "索引" document_type => "文档类型" template_overwrite => 是否重写 }
Codec插件介绍
1.plain 读取原始内容
2.rubydebug将Logstash Event按照ruby格式输出，方便调试
3.line处理带有换行符的内容
4.json处理json格式的内容
5.multiline处理多行数据的内容
七、最后说点什么
参考与学习的来源为慕课网和官方文档，欢迎大家加入我QQ群:438836709
欢迎大家关注我公众号:
另外Oath2.0的文章和demo正在努力写！！没事大家动动小手点个赞！！谢谢！！