用过Logstash的人都有个体会：内置插件再丰富，总有覆盖不到的场景。比如对方系统只认一套内部私有协议，官方自带的http和tcp插件都对接不上；或者数据进来之后要按公司内部的规则做脱敏和字段重组，硬凑grok加一堆filter配置也能跑，但规则一复杂，配置就变成了天书。这时候就得自己动手写Logstash自定义Java插件。而最稳的起点，不是从空的Maven工程开始搭框架，而是直接基于官方例子去改——官方生成器会把Gradle工程、SDK接口、配置文件全部铺好，你要做的只是在模板基础上替换自己的业务逻辑。

这篇文章直接讲操作：官方模板怎么生成、每个文件是干什么的、具体要改哪几处、改完怎么编译安装到Logstash里跑通，最后把我踩过的坑和排查方法一并列出来。准备写Java插件，或者正在跟官方文档较劲的朋友，可以按这个流程直接往下走。

1. 选型与准备：先搞明白你要动手改哪一类插件

1.1 五类插件，先知道自己要写的是哪一种

Logstash插件不是只有input和output两种，严格来说是五类：

• input：数据从外部进来，插件负责把外部数据拉进来并组装成事件，典型是读数据库、轮询HTTP接口、监听TCP端口。
• filter：事件在管道中间被改造，插件负责对已有事件做字段增删、格式转换、数据富化。
• output：事件最终要吐出去，插件负责把规范化事件写给下游系统。
• codec：负责对输入输出的数据做自定义编解码，比如私有二进制协议。
• pattern：grok模式扩展，严格来说就是给pattern文件加匹配规则，不写Java。

怎么快速判断自己该写哪一类？就看你的数据在管道里是"从哪一步参与的"：想从某处拿到原始数据，选input；想把某个字段从A变成B，选filter；想把最终事件写到某个地方，选output。一次只围绕一个类型来写，别上来就想搞一个大而全的插件包。官方模板也是按单类型生成的，混着写会让编译和排错的复杂度成倍增加。

1.2 用官方命令生成模板，比手搭工程靠谱得多

官方提供了一个插件生成器，在安装好的Logstash目录下直接执行：

bin/logstash-plugin generate --type input --name private_api --path /tmp/plugin-dev

--type 指定插件类型，--name 指定插件名（小写下划线风格），--path 是生成的目录。执行完之后，你会在 /tmp/plugin-dev 下得到一个可独立编译的Gradle工程。

我见过不少人放着这条官方路径不走，非要自己搭一个空Java工程再手动引入logstash-core依赖。结果也很统一：卡在版本匹配和SDK接口细节上。不同版本的Logstash接口是有差异的，光是把接口签名对齐就要折腾小半天。官方模板的价值恰恰在于，生成时SDK版本、Gradle配置、打包Task全部配套好了，你不需要去查"这个版本该用哪个接口"，直接在模板框架里填业务逻辑就行。

生成好的目录大致是这样的：

private_api/ ├── build.gradle ├── gradle.properties ├── settings.gradle ├── gradlew ├── gradlew.bat └── src/ ├── main/java/org/logstash/plugins/inputs/PrivateApi.java └── test/java/org/logstash/plugins/inputs/PrivateApiTest.java

拿到这个骨架之后，先别急着动手改代码，建议花几分钟把工程结构和模板读一遍。理解之后再动手，后面每一步都会很顺。

1.3 动手前先把你的需求翻译成数据流

改代码之前最容易犯的错误是：拿到模板就开始写循环，写到一半发现配置项少一个、字段名也没想清楚。我的习惯是先在一张纸上把数据流画清楚：

• 数据从哪来？格式是JSON、文本、二进制还是行协议？
• 每个事件要产生哪些字段？哪些必须、哪些可选？
• 需要用户在pipeline配置里设置哪些参数？给每个参数定好默认值和校验规则。
• 异常怎么处理？是重试、丢弃，还是把原始数据塞进event里交给下游兜底。
• 插件要不要支持多实例并发？如果多个pipeline或多个线程同时用到这个插件，内部状态怎么隔离。

这些问题想清楚，再去改代码，你会发现你只是在把上面的答案一行行落到模板的对应位置。这个习惯能帮你省掉至少一半的返工时间。

2. 官方模板逐文件拆解：先看懂再动手

2.1 build.gradle 和 gradle.properties：编译与打包的命脉

先看 build.gradle。官方模板的核心依赖是logstash-core：

dependencies { compileOnly "org.logstash:logstash-core:8.13.0" testImplementation "org.logstash:logstash-core:8.13.0" testImplementation "org.junit.jupiter:junit-jupiter:5.10.0" }

compileOnly 很关键：编译时需要这些类，但打gem包时必须排除，因为运行时Logstash自己已经带了SDK。如果你误改成implementation，最后gem体积会变大，运行时还可能和Logstash自带的jar产生类冲突。

gradle.properties 里主要放版本号和插件元信息。改插件名时这里要同步改，因为后面生成gem的文件名、安装时识别的插件名都跟它相关。

注意一个细节：官方生成器的模板，包路径默认是org.logstash.plugins加插件类型，如果你生成后要改包名，一定要把整个src/main/java下的目录结构一起改，不能只改代码里的package声明。

2.2 主类骨架：一个插件类是怎么被Logstash调起来的

以input模板为例，源码里最关键的内容是这几块。

第一，类的顶部有一个注解：

@LogstashPlugin(name = "private_api") public class PrivateApi implements Input

这个注解里的name，就是你在pipeline配置里写input { private_api { ... } }时用的插件名，必须小写、用下划线分隔。如果这里写错，装进Logstash后list命令能看到插件，但配置解析时会报找不到插件。

第二，配置项的声明方式。模板里会有一组静态常量，比如：

public static final PluginConfigSpec<Long> CONFIG_INTERVAL = PluginConfigSpec.numSetting("interval", 5L);

这行代码的意思是：插件接受一个名为 interval 的配置项，类型是数字，默认值是5。所有配置项都要在 configSchema() 方法里返回，Logstash才会在启动时帮你做类型校验和默认值注入。漏掉任何一个，这个配置项就永远解析不进插件里。

第三，生命周期方法。input插件有四个关键方法要看懂：

• start(InputContext)：插件启动入口，一般在这里提交一个线程做轮询或监听。
• run()：这是模板自己定义的私有方法，不是SDK接口，里面是主循环，负责源源不断产生Event并通过 eventConsumer.accept(event) 提交。
• stop()：收到停止信号时调用，通常置一个volatile布尔变量让循环退出。
• awaitStop()：阻塞等待插件完全停掉，通常配合CountDownLatch使用。

filter模板的核心逻辑更简单：实现 filter(Collection events, FilterMatchListener matchListener)，拿到一批事件，逐个改完再返回即可，不需要自己管理线程。output模板的核心是 write(Event event, OutputContext context)，每来一个事件调用一次。

2.3 测试目录和文档位：官方预留的扩展点

模板的test目录里有一个对应当前类名的测试文件。我刚开始接触时觉得它就是个摆设，后来发现这是最值得利用的部分。你不需要一上来就写完整测试，但至少保留一个能编译通过的测试类，这样每次改完代码运行gradlew test，能立刻知道有没有把骨架改坏。

README.md 和 CHANGELOG.md 也建议留着自己填。尤其是插件将来要共享给团队用的时候，配置项说明放在README里，能少回答很多群里重复的问题。

3. 动手改造：从官方例子到自己的插件

3.1 第一步：全局改名，别让插件名和类名打架

生成模板时已经指定了 --name，所以插件名和类名默认是匹配的。如果你生成时用了默认名，或者中途想改名，需要同步改这几个地方：

• @LogstashPlugin 注解里的 name
• Java 类名，以及对应的文件名
• 包名目录src/main/java/...下的路径结构
• build.gradle 里的插件元信息，比如 gemName、version 等
• 测试类里的同名类引用

如果不小心只改了注解没有改类名，编译能过，安装也能过，但后面排障会非常绕。我建议在生成模板之前就想好名字，不要在生成后再大改。

3.2 第二步：按需定义你自己的配置项

内置插件写配置时，如果配了一个不认识的参数，日志会提示 unknown setting。这个机制就是由 configSchema 控制的。自定义插件加配置项，写法是在类里声明静态常量，然后在 configSchema() 里返回。

常用的配置类型有这些：

• stringSetting("key", "默认值")：字符串
• numSetting("key", 1L)：数字
• booleanSetting("key", true)：布尔
• passwordSetting("key", "")：密码，日志里不会明文打印
• arraySetting("key", 默认数组)：数组

比如我要给私有API插件加上 endpoint 和 token：

public static final PluginConfigSpec<String> CONFIG_ENDPOINT = PluginConfigSpec.stringSetting("endpoint", "https://internal.example.com"); public static final PluginConfigSpec<String> CONFIG_TOKEN = PluginConfigSpec.stringSetting("token", ""); public static final PluginConfigSpec<Long> CONFIG_INTERVAL = PluginConfigSpec.numSetting("interval", 10L);

然后在构造函数里取值：

this.endpoint = config.get(CONFIG_ENDPOINT); this.token = config.get(CONFIG_TOKEN); this.interval = config.get(CONFIG_INTERVAL);

构造函数里可以做参数校验。比如endpoint必须以https开头，否则直接抛异常，Logstash会把这个异常编成配置错误信息展示出来，而不是等你跑到一半才炸。

还有一个细节：如果某配置项必须由用户显式指定，不带默认值，可以用 PluginConfigSpec.requiredSetting 这类方法生成。具体方法名在你用的SDK版本里会有略微差异，以模板里提示的为准。

3.3 第三步：把模板的假逻辑换成你的真逻辑

这是整个改造的核心。input模板默认的run方法只是每隔几秒产生一条 hello 事件，你要把它替换成你自己的数据获取和事件构造逻辑。

比如对接一个内部HTTP接口，每10秒拉一次JSON列表，每条记录生成一个Event：

package org.logstash.plugins.inputs; import co.elastic.logstash.api.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Arrays; import java.util.Collection; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @LogstashPlugin(name = "private_api") public class PrivateApi implements Input { public static final PluginConfigSpec<String> CONFIG_ENDPOINT = PluginConfigSpec.stringSetting("endpoint", "https://internal.example.com"); public static final PluginConfigSpec<String> CONFIG_TOKEN = PluginConfigSpec.stringSetting("token", ""); public static final PluginConfigSpec<Long> CONFIG_INTERVAL = PluginConfigSpec.numSetting("interval", 10L); private final String id; private final String endpoint; private final String token; private final Long interval; private final Logger logger = LoggerFactory.getLogger(PrivateApi.class); private final CountDownLatch done = new CountDownLatch(1); private volatile boolean stopped; private ExecutorService executor; private EventConsumer<Event> eventConsumer; public PrivateApi(String id, Configuration config, Context context) { this.id = id; this.endpoint = config.get(CONFIG_ENDPOINT); this.token = config.get(CONFIG_TOKEN); this.interval = config.get(CONFIG_INTERVAL); } @Override public void start(InputContext context) { this.eventConsumer = context.getEventConsumer(); executor = Executors.newSingleThreadExecutor(); executor.submit(this::run); } private void run() { while (!stopped) { try { List<Map<String, Object>> items = fetchItems(endpoint, token); for (Map<String, Object> item : items) { Event event = new Event(); for (Map.Entry<String, Object> entry : item.entrySet()) { event.setField(entry.getKey(), entry.getValue()); } event.setField("source", "private_api"); eventConsumer.accept(event); } } catch (Exception e) { logger.warn("拉取数据失败，等待下个周期重试", e); } try { Thread.sleep(interval * 1000L); } catch (InterruptedException e) { stopped = true; } } done.countDown(); } @Override public void stop() { stopped = true; executor.shutdownNow(); } @Override public void awaitStop() throws InterruptedException { done.await(); } @Override public Collection<PluginConfigSpec<?>> configSchema() { return Arrays.asList(CONFIG_ENDPOINT, CONFIG_TOKEN, CONFIG_INTERVAL); } @Override public String getId() { return id; } }

注意我在代码里没有手动设置 @timestamp，因为Logstash会自动给事件补充这个字段。你只需要关注业务字段，不要刻意覆盖系统字段，否则在Kibana里排序会出现意料之外的情况。

filter插件类似，模板里往往只是原样返回事件，你要做的是在filter方法里对事件做修改。比如一个脱敏逻辑：

@Override public Collection<Event> filter(Collection<Event> events, FilterMatchListener matchListener) { for (Event event : events) { Object phone = event.getField("phone"); if (phone instanceof String && ((String) phone).length() == 11) { String masked = ((String) phone).replaceAll("(\\d{3})\\d{4}(\\d{4})", "$1****$2"); event.setField("phone", masked); event.setField("masked_flag", true); } } return events; }

filter返回的集合就是后续管道继续处理的集合。如果你把某个事件从返回集合里移除，就相当于丢掉了这条事件。这个特性可以用来做过滤和采样，但也要小心：当你把事件移除时，管道后面所有插件都看不到它了。

output插件则是在write方法里把event的字段取出来，按照目标系统的协议发送过去。整体思路跟input是镜像关系。

3.4 别忘了生命周期：stop和awaitStop要配合好

这句话强调多少遍都不为过：input插件的run循环，如果没有正确的停止机制，Logstash关闭时会卡住。标准写法就是模板里那样：用volatile开关，stop()里把布尔置true，循环在sleep被InterruptedException打断后退出。

如果你自己起了线程池，记得在stop或awaitStop里调用 executor.shutdownNow()，避免线程泄漏。实测下来，开关变量加打断sleep这种组合，在绝大多数场景下都比复杂并发工具更可靠。filter和output插件没有run循环，生命周期处理相对简单，只需要在stop里释放连接之类的资源就行。

4. 编译、安装、联调：把插件真正跑起来

4.1 本地编译：先过Gradle这一关

进入插件目录，执行：

./gradlew clean build

正常会跑完编译、测试、打包，在 build/libs/ 下生成一个 .gem 文件，文件名一般是 logstash-input-private_api-0.1.0.gem 这种风格。第一次跑Gradle会下载依赖，网络一般的话耐心等一会儿。

最容易踩到的编译问题是JDK版本。Logstash 8.x 对应 JDK 17，如果你本机默认JDK是8或11，大概率会编译报错。可以用./gradlew -version查看当前Gradle使用的JDK版本，如果不对，在 gradle.properties 里配置org.gradle.java.home指向正确的JDK目录，或者直接切换环境变量。

4.2 把插件装进Logstash并验证

安装本地gem的命令是：

bin/logstash-plugin install --local /path/to/build/libs/logstash-input-private_api-0.1.0.gem

装完用 list 确认一下：

bin/logstash-plugin list | grep private_api

看到列表里出现 logstash-input-private_api 就走通了第一步。接下来写一个最简pipeline验证功能：

input { private_api { endpoint => "https://internal.example.com/api/list" token => "xxxx" interval => 10 } } output { stdout { codec => rubydebug } }

用bin/logstash -f test-pipeline.conf启动，观察stdout输出。如果没有任何输出，先看管道日志，再看input有没有打印异常。

配置错误时的表现很有诊断价值：插件自己抛出的异常会出现在Logstash启动日志的配置校验阶段。如果连配置阶段都没报错，说明插件名和配置项签名已经匹配上了，问题大概率出在业务逻辑里。

4.3 联调时怎么利用好日志

Logstash的Java插件日志基于SLF4J，模板里一般已经在类顶部声明了logger：

private final Logger logger = LoggerFactory.getLogger(PrivateApi.class);

你可以在关键位置插日志：start时打印配置项，run循环里打印拉取条数，catch里打印完整堆栈。配合 Logstash 的 log.level: debug（在logstash.yml里配置，或启动时加 --debug），排障效率会高很多。

还有一个很实用的技巧：先用官方插件作为参照物。比如你的场景是轮询HTTP接口，先用自带的 http_poller 确认接口数据能被正常拿到，再把同样的参数迁移到自定义插件里。这样能快速把问题边界缩小到"是我的插件逻辑问题"还是"接口本身问题"。

5. 我踩过的那些坑：问题速查与排错心得

5.1 高发问题与解决对照表

这里把开发Custom Java插件过程中最常遇到的问题整理成一张速查表，留着备用：

第二行的"gem文件名不对"是我自己实际遇到过的。因为官方生成器默认生成的gemName可能不带插件类型前缀，如果你把build.gradle里的元信息改乱了，生成的.gem可能无法被Logstash识别。修复方案就是对照官方模板里的gradle配置，按 rule 填好即可。

5.2 两个提高开发效率的实践建议

第一，先串通最小闭环再加逻辑。很多人习惯一次性把插件写得很完整，结果编译通过后跑起来一堆问题。我现在的做法反过来了：先用模板默认逻辑生成一条事件，把pipeline从input到output整条串通，确认插件能被安装、事件能流到stdout之后，再逐步增加真实业务代码。每一步改动都能快速定位问题，试错成本低很多。

第二，测试代码别删。模板自带的单测虽然简单，但它是验证SDK接口用法的活例子。改造过程中如果怀疑某个接口调用不对，直接在测试里跑一个最小方法是最快的验证方式。把gradlew test养成日常习惯，比手工启动整个Logstash去验证快得多。

最后说一点个人体会。一开始写自定义插件时，我也试图从零搭建工程、自己管理依赖，后来发现最省时省力的方式就是老老实实从官方模板改起。官方模板的价值不仅在于省掉配置工程的功夫，更在于它展示了一套Logstash期望插件长什么样的标准姿势：配置项怎么声明、生命周期怎么配合、事件怎么提交。照着这个姿势改，你的插件在行为上就和内置插件一致，装上去、配起来、调起来都顺。

如果你正准备改造一个插件，我的建议是：先花半小时把模板读懂，再开始改。代码量真不大，读懂了后面每一步都很快。跑通第一个插件之后，你会发现后面再写filter、output，基本就是同样的套路，只是核心方法名换了一下而已。