企业官网建设流程全解析

• 文档
• 教程
• 数据库

本文聚焦 Elastic 技术栈（ELK）中的 Logstash，系统讲解其定位、三阶段处理管道（输入 > 过滤器 > 输出）、设置文件与关键配置项、启动方式、插件体系，并结合仓库内提供的实战案例，给出「控制台数据」「logback 日志（TCP/UDP）」「日志文件」三类典型场景的可运行配置。读完本文，你将能够独立编写 Logstash 管道配置，将各类数据源接入 Elasticsearch，并在 Linux 环境下以守护进程方式稳定运行 Logstash。

如果对 Elastic 的安装、配置、部署还不熟悉，可以参考本仓库的 Elastic 快速入门；同一技术栈下的 Filebeat 与 Kibana 文档同样值得阅读。

简介

Logstash 可以传输和处理你的日志、事务或其他数据，是 Elasticsearch 的最佳数据管道。

Logstash 采用插件式管理模式，在输入、过滤、输出以及编码（codec）过程中都可以使用插件进行定制。Logstash 社区提供了超过 200 种可用插件，覆盖了绝大多数常见数据源与数据处理需求。

工作原理

Logstash 管道（pipeline）有两个必要元素：input和output，以及一个可选元素：filter。

这三个元素分别代表 Logstash 事件处理的三个阶段：输入（input）> 过滤器（filter）> 输出（output）。

• input：负责从数据源采集数据；
• filter：将数据修改为你指定的格式或内容；
• output：将数据传输到目的地。

在实际应用场景中，通常输入、输出、过滤器不止一个。Logstash 的这三个元素都使用插件式管理方式，用户可以根据应用需要，灵活选用各阶段需要的插件并组合使用。例如，可以在一个管道中同时监听多个文件、同时向 Elasticsearch 与文件系统输出。

设置

设置文件

Logstash 安装后主要包含三个设置文件：

• logstash.yml：Logstash 的默认启动配置文件，控制节点名、管道参数、日志级别等全局行为；
• jvm.options：Logstash 的 JVM 配置文件，用于调整堆内存等 JVM 参数；
• startup.options（Linux）：包含系统安装脚本在/usr/share/logstash/bin中使用的选项，用于为您的系统构建适当的启动脚本。安装 Logstash 软件包时，系统安装脚本会在安装过程结束时执行，并使用startup.options中指定的设置来设置用户、组、服务名称和服务描述等选项。

若需要安装与运行层面的进一步说明（下载解压、最小可运行配置、Windows 下bin/logstash.bat -f logstash.conf的启动方式等），可参考仓库中的 Logstash 运维 文档。

logstash.yml 设置项

下表节选了logstash.yml中的部分核心设置项，更多项请参考 Logstash 官方文档的 logstash-settings-file 章节。

几个值得重点关注的调优要点：

• pipeline.batch.size与pipeline.batch.delay共同决定单批事件的构成方式：前者是数量上限，后者是凑批等待时间。增大批量通常能提升吞吐，但会抬高内存占用，必要时需配合LS_HEAP_SIZE（或jvm.options中的-Xmx）调大堆内存；
• config.reload.automatic配合config.reload.interval可实现配置热加载，适合线上调试管道而不中断服务；
• config.test_and_exit可用于 CI 场景：启动前先校验管道配置合法性，避免带病上线。

启动

命令行

通过命令行启动 Logstash 的方式如下：

bin/logstash [options]

其中options是您可以指定用于控制 Logstash 执行的命令行标志。在命令行上设置的任何标志都会覆盖 Logstash 设置文件（logstash.yml）中的相应设置，但设置文件本身不会更改。

注

虽然可以通过指定命令行参数的方式控制 Logstash 的运行方式，但显然这么做很麻烦。

建议通过指定配置文件的方式控制 Logstash 运行，启动命令如下：

bin/logstash -f logstash.conf

若想了解更多的命令行参数细节，请参考 Logstash 官方文档的 running-logstash-command-line 章节。

配置文件

上节我们了解到，Logstash 可以执行bin/logstash -f logstash.conf，按照配置文件中的参数去覆盖默认设置文件（logstash.yml）中的设置。本节我们来学习这个配置文件如何配置参数。

配置文件结构

在工作原理一节中，我们已经知道 Logstash 主要有三个工作阶段 input、filter、output，而 Logstash 配置文件结构也与之相对应：

input {} filter {} output {}

每个部分都包含一个或多个插件的配置选项。如果指定了多个过滤器，则会按照它们在配置文件中的显示顺序应用它们。

插件配置

插件的配置由插件名称和插件的一个设置块组成。下面的例子中配置了两个 file 输入插件：

input { file { path => "/var/log/messages" type => "syslog" } file { path => "/var/log/apache/access.log" type => "apache" } }

您可以配置的设置因插件类型而异，具体可参考 Input Plugins、Output Plugins、Filter Plugins 和 Codec Plugins 等官方插件文档。从本仓库的 Logstash 运维 文档可以看到，多个同类型插件（如两个file）可以在同一个input块内并列声明，每个插件实例独立采集各自的数据源。

值类型

一个插件可以要求设置的值是一个特定的类型，比如布尔值、列表或哈希值。以下值类型受支持。

• Array

users => [ {id => 1, name => bob}, {id => 2, name => jane} ]

• Lists

path => ['/var/log/messages', '/var/log/*.log'] uris => ['http://elastic.co', 'http://example.net']

• Boolean

ssl_enable => true

• Bytes（支持字节单位后缀，MiB/kib/mb等均可用）

my_bytes => "1113" # 1113 bytes my_bytes => "10MiB" # 10485760 bytes my_bytes => "100kib" # 102400 bytes my_bytes => "180 mb" # 180000000 bytes

• Codec

codec => 'json'

• Hash

match => { "field1" => "value1" "field2" => "value2" ... }

• Number

port => 33

• Password

my_password => 'password'

• URI

my_uri => 'http://foo:bar@example.net'

• Path

my_path => '/tmp/logstash'

• String

my_string => "hello world"

• 转义字符：当config.support_escapes设置为 true 时，带引号的字符串将处理转义字符（如\n、\t等），否则这些字符会被原样保留。

配置语法要点小结：key => value是插件设置的标准写法；设置值可以是字符串、数字、布尔值、数组、哈希等多种类型；字节类型的值支持人类可读的单位后缀，便于直接书写10MiB这样的配置。

插件

input（输入插件）

Logstash 支持各种输入选择，可以在同一时间从众多常用来源捕捉事件，能够以连续的流式传输方式，轻松地从您的日志、指标、Web 应用、数据存储以及各种 AWS 服务采集数据。

常用 input 插件

• file：从文件系统上的文件读取，就像 UNIX 命令tail -0F一样，持续跟踪文件的增量内容；
• syslog：在众所周知的端口 514 上侦听系统日志消息，并根据 RFC3164 格式进行解析；
• redis：从 redis 服务器读取，使用 redis 通道和 redis 列表。Redis 经常用作集中式 Logstash 安装中的“代理”，它将来自远程 Logstash“托运人”的 Logstash 事件排队，起到削峰填谷的作用；
• beats：处理由 Filebeat 发送的事件。Filebeat 的配置示例（output.logstash指向 Logstash 的5044端口，Logstash 侧用beats { port => 5044 }接收）可以参考仓库中的 Elastic 技术栈之 Filebeat 文档。

更多详情请见 Input Plugins 官方文档。

filter（过滤器）

过滤器是 Logstash 管道中的中间处理设备。如果符合特定条件，您可以将条件过滤器组合在一起，对事件执行操作。

常用 filter 插件

• grok：解析和结构化任意文本。Grok 目前是 Logstash 中将非结构化日志数据解析为结构化和可查询数据的最佳方法；
• mutate：对事件字段执行一般转换。您可以重命名、删除、替换和修改事件中的字段；
• drop：完全放弃一个事件，例如丢弃调试事件；
• clone：制作一个事件的副本，可能会添加或删除字段；
• geoip：添加有关 IP 地址的地理位置的信息（也可以在 Kibana 中显示图表）。

更多详情请见 Filter Plugins 官方文档。

output（输出插件）

输出是 Logstash 管道的最后阶段。一个事件可以通过多个输出，但是一旦所有输出处理完成，事件就完成了执行。

常用 output 插件

• elasticsearch：将事件数据发送给 Elasticsearch（推荐模式）；
• file：将事件数据写入文件或磁盘；
• graphite：将事件数据发送给 graphite（一个流行的开源工具，用于存储和绘制指标）；
• statsd：将事件数据发送到 statsd（这是一种侦听统计数据（如计数器和定时器）的服务，通过 UDP 发送并将聚合发送到一个或多个可插入的后端服务）。

更多详情请见 Output Plugins 官方文档。

codec（编解码插件）

codec 用于格式化对应的内容，即处理事件在传输过程中的编码与解码。

常用 codec 插件

• json：以 JSON 格式对数据进行编码或解码；
• multiline：将多行文本事件（如 Java 异常和堆栈跟踪消息）合并为单个事件。

在实战中，codec => json_lines、codec => json、codec => rubydebug会频繁出现：json_lines用于解析按行分隔的 JSON 流（logback TCP 场景），json用于解析单条 JSON 消息（UDP 场景），rubydebug则用于在标准输出上以可读的 Ruby 风格结构打印事件，方便调试。

更多插件请见 Codec Plugins 官方文档。

实战

前面的内容都是对 Logstash 的介绍和原理说明。接下来，我们来实战一些常见的应用场景。

传输控制台数据

stdin input 插件从标准输入读取事件，这是最简单的 input 插件，一般用于测试场景。

应用

（1）创建logstash-input-stdin.conf：

input { stdin { } } output { elasticsearch { hosts => ["localhost:9200"] } stdout { codec => rubydebug } }

（2）执行 Logstash，使用-f来指定你的配置文件：

bin/logstash -f logstash-input-stdin.conf

启动后，在终端输入任意文本并回车，该事件会被写入 Elasticsearch（localhost:9200），同时以rubydebug格式打印到终端，是验证管道连通性的最快方式。

传输 logback 日志

ELK 默认使用的 Java 日志工具是 log4j2，并不支持 logback 和 log4j。

想使用 logback + logstash，可以使用 logstash-logback-encoder。该库提供了 UDP / TCP / 异步方式来传输日志数据到 Logstash。

如果你使用的是 log4j，也可以用这种方式，只要引入桥接 jar 包即可。如果你对 log4j、logback 或是桥接 jar 包不太了解，可以参考相关 Java 日志工具库的博文。

TCP 应用

logstash 配置

（1）创建logstash-input-tcp.conf：

input { tcp { port => 9251 codec => json_lines mode => server } } output { elasticsearch { hosts => ["localhost:9200"] } stdout { codec => rubydebug } }

这里把 Logstash 作为 TCP 服务端（mode => server），在9251端口接收 Java 应用推送过来的、以换行分隔的 JSON 日志流（codec => json_lines）。

（2）执行 Logstash，使用-f来指定你的配置文件：bin/logstash -f logstash-input-tcp.conf

java 应用配置

（1）在 Java 应用的 pom.xml 中引入 jar 包：

<dependency> <groupId>net.logstash.logback</groupId> <artifactId>logstash-logback-encoder</artifactId> <version>4.11</version> </dependency> <!-- logback 依赖包 --> <dependency> <groupId>ch.qos.logback</groupId> <artifactId>logback-core</artifactId> <version>1.2.3</version> </dependency> <dependency> <groupId>ch.qos.logback</groupId> <artifactId>logback-classic</artifactId> <version>1.2.3</version> </dependency> <dependency> <groupId>ch.qos.logback</groupId> <artifactId>logback-access</artifactId> <version>1.2.3</version> </dependency>

（2）接着，在 logback.xml 中添加 appender：

<appender name="ELK-TCP" class="net.logstash.logback.appender.LogstashTcpSocketAppender"> <!-- destination 是 logstash 服务的 host:port， 相当于和 logstash 建立了管道，将日志数据定向传输到 logstash --> <destination>192.168.28.32:9251</destination> <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder"/> </appender> <logger name="io.github.dunwu.spring" level="TRACE" additivity="false"> <appender-ref ref="ELK-TCP" /> </logger>

（3）接下来就是 logback 的具体使用。此后，io.github.dunwu.spring包中的 TRACE 及以上级别的日志信息都会被定向输出到 Logstash 服务，经管道处理后写入 Elasticsearch。

注意：示例中的destination端口（9251）必须与logstash-input-tcp.conf中tcp { port => 9251 }保持一致；192.168.28.32为示例 Logstash 主机地址，请替换为你自己的环境。

UDP 应用

UDP 和 TCP 的使用方式大同小异。

logstash 配置

（1）创建logstash-input-udp.conf：

input { udp { port => 9250 codec => json } } output { elasticsearch { hosts => ["localhost:9200"] } stdout { codec => rubydebug } }

（2）执行 Logstash，使用-f来指定你的配置文件：bin/logstash -f logstash-input-udp.conf

java 应用配置

（1）在 Java 应用的 pom.xml 中引入 jar 包：与TCP 应用一节中的引入依赖包完全相同。

（2）接着，在 logback.xml 中添加 appender：

<appender name="ELK-UDP" class="net.logstash.logback.appender.LogstashSocketAppender"> <host>192.168.28.32</host> <port>9250</port> </appender> <logger name="io.github.dunwu.spring" level="TRACE" additivity="false"> <appender-ref ref="ELK-UDP" /> </logger>

（3）接下来就是 logback 的具体使用。

TCP 与 UDP 场景的差异集中在三处：Logstash 侧插件由tcp换成udp、codec 由json_lines换成json；Java 侧 appender 由LogstashTcpSocketAppender换成LogstashSocketAppender；端口需与 Logstash 配置一一对应（示例中 TCP 用9251、UDP 用9250）。UDP 面向日志场景可以容忍偶发丢包以换取更低的网络开销，TCP 则保证可靠传输，请按业务对可靠性的要求取舍。

传输文件

在 Java Web 领域，需要用到一些重要的工具，例如 Tomcat、Nginx、MySQL 等。这些不属于业务应用，但是它们的日志数据对于定位问题、分析统计同样很重要，这时无法使用 logback 方式将它们的日志传输到 Logstash。

如何采集这些日志文件呢？可以使用 Logstash 的 file input 插件。

需要注意的是，传输文件这种方式，必须在日志所在的机器上部署 Logstash。

应用

logstash 配置

（1）创建logstash-input-file.conf：

input { file { path => ["/var/log/nginx/access.log"] type => "nginx-access-log" start_position => "beginning" } } output { if [type] == "nginx-access-log" { elasticsearch { hosts => ["localhost:9200"] index => "nginx-access-log" } } }

（2）执行 Logstash，使用-f来指定你的配置文件：bin/logstash -f logstash-input-file.conf

说明：

• path支持数组写法，可同时跟踪多个文件或通配符匹配的日志集合；
• type为事件打上类型标签，输出阶段用if [type] == "..."条件判断实现按类型路由；
• start_position => "beginning"表示从文件头开始读取（适用于首次接入的历史日志）；默认情况下 file 插件会记录每个文件的读取位置（sincedb），重启后从上次断点继续，避免重复消费；
• index指定写入 Elasticsearch 的目标索引名，方便后续在 Kibana 中按索引检索与分析。

小技巧

启动、终止应用

如果你的 Logstash 每次都是通过指定配置文件方式启动，不妨建立一个启动脚本：

# cd xxx 进入 logstash 安装目录下的 bin 目录 logstash -f logstash.conf

如果你的 Logstash 运行在 Linux 系统下，不妨使用 nohup 来启动一个守护进程。这样做的好处在于，即使关闭终端，应用仍会运行。

创建 startup.sh

nohup ./logstash -f logstash.conf >> nohup.out 2>&1 &

• nohup使进程忽略挂断信号，终端关闭后进程继续运行；
• >> nohup.out 2>&1将标准输出与标准错误重定向到nohup.out，方便事后排查启动问题；
• 末尾的&让命令在后台执行。

终止应用没有什么好方法，你只能使用ps -ef | grep logstash查出进程，将其 kill。不过，我们可以写一个脚本来干这件事：

创建 shutdown.sh

脚本不多解释，请自行领会作用。

PID=`ps -ef | grep logstash | awk '{ print $2}' | head -n 1` kill -9 ${PID}

提示：kill -9属于强杀。在 Logstash 中，如果设置了pipeline.unsafe_shutdown => false（默认），优雅关闭会等待内存中所有事件推送完毕；使用kill -9则可能丢失 inflight 事件。生产环境建议优先尝试kill ${PID}（默认 SIGTERM），确实无响应时再考虑强杀。

资料

• Logstash 官方文档
• logstash-logback-encoder
• ELK Stack 权威指南
• ELK（Elasticsearch、Logstash、Kibana）安装和配置

推荐阅读

• Elastic 快速入门
• Elastic 技术栈之 Filebeat
• Elastic 技术栈之 Kibana
• Logstash 运维

• 文档
• 教程
• 数据库