Debezium日常分享系列之：提升Debezium性能

定期评估整个项目（或至少其关键部分）的性能表现十分必要，尤其是在新增功能或进行重大代码重构时。不过，性能检查也可以临时开展，或更理想地——形成定期机制。

本文将通过一个案例，演示如何快速识别和分析Debezium中的特定性能问题。内容涵盖完整闭环：搭建轻量级性能测试、分析结果、提出优化方案并验证效果。

建立完善的性能测试通常充满挑战，许多细节可能导致结果失真。本文介绍的方法旨在快速暴露潜在瓶颈——通常是那些通过简单测试就能发现的"低垂果实"。当然，这并不意味着常规性能测试不重要。恰恰相反，全面且周期性的测试（尤其是使用高级负载生成工具的端到端测试）至关重要，许多性能问题只有在复杂环境中才会显现。

测试环境搭建

Debezium应用程序

我们首先基于Debezium嵌入式引擎和PostgreSQL连接器创建一个简单应用。该应用采用极简配置，主要依赖默认设置。特别说明的是，它使用decoderbufs插件从PostgreSQL数据库获取数据。

应用本身不处理获取的数据——这是为了刻意避免任何记录后处理（例如单消息转换）。这种设计确保应用耗时集中在核心数据获取路径上，而非外部处理逻辑。换言之，我们的重点是评估通过decoderbufs插件从PostgreSQL检索数据的效率，并定位该路径中的潜在性能瓶颈。

public class DebeziumEngniePostgres {public static void main(String[] args) {final Properties props = new Properties();props.setProperty("name", "engine");props.setProperty("connector.class", "io.debezium.connector.postgresql.PostgresConnector");props.setProperty("database.hostname", "127.0.0.1");props.setProperty("database.port", "5432");props.setProperty("database.user", "postgres");props.setProperty("database.password", "postgres");props.setProperty("database.dbname", "postgres");props.setProperty("topic.prefix", "perf");props.setProperty("table.include.list", "public.pgbench_.*");props.setProperty("snapshot.mode", "no_data");props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");props.setProperty("offset.storage.file.filename", "./data/offsets.dat");props.setProperty("offset.flush.interval.ms", "60000");props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory");props.setProperty("schema.history.internal.file.filename", "./data/schemahistory.dat");try (DebeziumEngine<ChangeEvent<SourceRecord, SourceRecord>> engine = DebeziumEngine.create(Connect.class).using(props).notifying(record -> {}).build()) {ExecutorService executor = Executors.newSingleThreadExecutor();executor.execute(engine);System.out.println("Debezium started");Thread.sleep(600000);}catch (IOException|InterruptedException e) {System.out.println("Failed with " + e);}System.out.println("Debezium stopped");}
}

源码仓库地址

您可以像往常一样使用Maven构建和运行应用程序。只需首先创建数据目录，因为Debezium引擎会尝试将偏移量和模式历史文件存储在那里。

PostgreSQL

由于我们在这个示例中选择了PostgreSQL连接器，因此需要设置一个测试用的PostgreSQL数据库。为了简化操作，我们可以使用Debezium项目提供的容器镜像——它已经包含了所有必要的数据库配置，开箱即用：

podman run --rm --name postgres -it -e POSTGRES_PASSWORD=postgres -p 5432:5432 quay.io/debezium/example-postgres:3.2

接下来，我们需要一个工具在测试期间为数据库生成负载。PostgreSQL自带这样的工具：pgbench。在大多数Linux发行版中，它与核心数据库软件包是分开提供的。例如在Fedora系统上，您需要额外安装postgresql-contrib包。

建立常规性能测试时，关键环节之一是选择合适的负载生成工具。像pgbench这类工具可能存在诸如协调遗漏问题等缺陷，导致生成的数据不真实，或引入其他偏差从而产生误导性结果。如前所述，本文暂不探讨这些复杂问题——这是本示例中采用的简化处理方式之一，特此说明。

在使用pgbench之前，需要先初始化它将操作的测试表：

PGPASSWORD=postgres pgbench -h 127.0.0.1 -U postgres -i postgres --scale=10

该命令会在数据库中创建若干以pgbench_开头的表。您可能已经注意到，我们的应用程序已配置为捕获这些pgbench_*表的变更。

运行测试

现在一切准备就绪，让我们使用pgbench对数据库施加负载：

PGPASSWORD=postgres pgbench -h 127.0.0.1 -U postgres --scale=10 -b simple-update --jobs=20 --client=20 -T 120 postgres

同时，我们将运行启用 Java Flight Recorder 的测试应用程序：

java -XX:+FlightRecorder -XX:StartFlightRecording=delay=30s,duration=60s,filename=dbz-flight.jfr,settings=profile -jar dbz-app/target/debezium-quick-perf-1.0-SNAPSHOT.jar

要捕获Flight Recorder事件，请确保在启动应用程序时包含-XX:+FlightRecorder参数。您可以通过性能分析工具手动开始记录，或通过命令行参数配置延迟时间、持续时间、输出文件和性能分析设置：-XX:StartFlightRecording=delay=30s,duration=60s,filename=dbz-flight.jfr,settings=profile.

当记录完成后，您可以同时停止应用程序和pgbench（如果它仍在运行）。也可以停止并删除数据库容器。

如需重复测试，建议从干净的数据库开始，并删除数据目录中的offset文件以重置连接器的读取位置。

分析结果

获取Flight Recorder文件后，我们可以用任何支持Java Flight Recorder (JFR)的工具打开它。这可以是您的IDE，但我更推荐使用专用工具：Java Mission Control (JMC)。

不过需要注意，JMC对Wayland的支持存在缺陷。这可能导致某些关键视图（如我们需要的火焰图）显示为空。详情可参阅JMC-8247问题记录。

火焰图能直观展示应用程序的时间消耗分布：

有一件事立即引人注目，那就是在 schemaChanged() 方法中花费了大量的时间，特别是在正则表达式匹配中：

同样的问题在方法分析视图中也很明显：

这里，正则表达式匹配是最耗时的操作之一。您还可以检查采样事件的调用堆栈，精确定位这个计算的来源。

问题的根源在于PostgreSQL不会发出专门的模式变更事件。相反，模式元数据被嵌入到第一个使用更新后模式的记录中。因此，Debezium必须几乎在处理的每条记录中检查可能的模式变更。具体实现取决于所使用的插件。对于默认的decoderbufs插件，Debezium会将记录的类型和其他修饰符与已有的信息进行比较。这些类型修饰符以字符串形式提供，并且每条记录都需要使用正则表达式进行解析。随着时间的推移，这会累积成很大的开销。

既然我们已经发现了一个潜在的性能瓶颈，最好将我们的发现报告给开发团队。针对这个具体问题，我提交了DBZ-9093。

提出修复方案

在上一步中，我们发现了PostgreSQL连接器在使用decoderbufs插件时存在的性能问题。幸运的是，当使用pgoutput插件时不存在这个问题。不过考虑到decoderbufs仍受支持且被设为默认选项，让我们探讨如何修复这个问题。

JMH基准测试

为了衡量修复效果，我们可以使用专为此目的设计的Java微基准测试工具(JMH)。一个简单的JMH基准测试示例如下：

@State(Scope.Benchmark)
public class PostgresTypeMetadataPerf {private static final int OP_COUNT = 10;private static final int MOD_COUNT = 10;private static final String[] MODIFIERS = {"text","character varying(255)","numeric(12,3)","geometry(MultiPolygon,4326)","timestamp (12) with time zone","int[]","myschema.geometry","float[10]","date","bytea"};private ReplicationMessage.Column[] columns = new ReplicationMessage.Column[OP_COUNT];private ReplicationMessage.Column createColumn (int modifierIndex) {String columnName = "test";PostgresType columnType = PostgresType.UNKNOWN;String typeWithModifiers = MODIFIERS[modifierIndex];boolean optional = true;return new AbstractReplicationMessageColumn(columnName, columnType, typeWithModifiers, optional) {@Overridepublic Object getValue(PostgresStreamingChangeEventSource.PgConnectionSupplier connection,boolean includeUnknownDatatypes) {return null;}};}@Setup(Level.Invocation)public void setup() {Random random = new Random(1234);for (int i = 0; i < OP_COUNT; i++) {columns[i] = createColumn(random.nextInt(MOD_COUNT));}}@Benchmark@BenchmarkMode(Mode.AverageTime)@OutputTimeUnit(TimeUnit.MICROSECONDS)@Fork(value = 1)@OperationsPerInvocation(OP_COUNT)public void columnMetadata(Blackhole bh) {for (int i = 0; i < OP_COUNT; i++) {bh.consume(columns[i].getTypeMetadata());}}
}

虽然 JMH 和微基准测试通常存在许多需要注意的事项，但这个基准测试至少为我们提供了一些基准。

以下是在我的机器上使用未打补丁的代码得出的结果：

Iteration   1: 0.768 us/op
Iteration   2: 0.761 us/op
Iteration   3: 0.780 us/op
Iteration   4: 0.780 us/op
Iteration   5: 0.750 us/opBenchmark                                Mode  Cnt  Score   Error  Units
PostgresTypeMetadataPerf.columnMetadata  avgt    5  0.768 ? 0.049  us/op

鉴于类型修饰符通常会被复用，一种简单的优化方法是将已解析的修饰符缓存到 map 中。实现此缓存逻辑后，JMH 基准测试给出了以下结果：

Iteration   1: 0.278 us/op
Iteration   2: 0.278 us/op
Iteration   3: 0.284 us/op
Iteration   4: 0.288 us/op
Iteration   5: 0.291 us/opBenchmark                                Mode  Cnt  Score   Error  Units
PostgresTypeMetadataPerf.columnMetadata  avgt    5  0.284 ? 0.023  us/op

虽然这是一个简单的基准测试（可能存在某些问题），但性能提升幅度已经足以表明实际应用中的收益。

在测试应用中进行验证

作为最终测试，我们可以重新运行简单的性能测试，检查schemaChanged()方法的时间消耗。从更新后的火焰图可以明显看出改进效果：

当您对修复方案有信心后，可以提交一个拉取请求——最好包含用于验证的JMH基准测试。

要高度确信该修复的有效性，需要建立完整的端到端性能测试流程。不过在如此复杂的环境中，实际效果可能微乎其微。与我们的受控测试或JMH基准测试不同，真实场景包含序列化、I/O开销等诸多因素，这些都可能稀释可见的收益。孤立测试中显著的性能提升，在整体上可能只带来微小改进——但积少成多总是好的。

结论

本文探索了一种轻量级方法来识别Debezium（使用PostgreSQL连接器的嵌入式引擎）中的性能瓶颈。通过简单的测试设置、Flight Recorder和火焰图分析，我们定位到了decoderbufs插件中正则表达式处理的高成本问题。

虽然这个设置有意简化了现实场景的许多方面，也存在其他潜在缺陷，但它对于发现容易解决的性能问题仍然有效。开发者可以直接在本地运行轻量级基准测试，有助于在开发早期发现并解决低效问题。这些优化能提升整体吞吐量并减少不必要的CPU使用，尤其在高吞吐场景中。虽然这种方法不能替代全面的端到端性能测试，但它提供了一种快速实用的方式，在开发过程中发现并解决性能退步或低效问题。在投入时间建立更复杂的基准测试流程之前，这也是一个很好的第一步。