You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Camel Quarkus SFTP幂等消费者启动报错：this.transactionTemplate为null

avatar
算力阿峰
2026-6-27
问题描述

我正在开发一个带有SFTP路由的Quarkus应用，希望集成IdempotentConsumer以在集群多实例部署时实现分布式锁。但应用启动时出现空指针错误，提示Cannot invoke "org.springframework.transaction.support.TransactionTemplate.execute(org.springframework.transaction.support.TransactionCallback)" because "this.transactionTemplate" is null。数据库连接正常（Liquibase日志已验证），Hibernate实体及EntityManager注入工作正常。

相关配置

pom.xml关键配置 

<!--pom.xml-->
<properties>
  <quarkus.platform.artifact-id>quarkus-bom</quarkus.platform.artifact-id>
  <quarkus.platform.group-id>io.quarkus.platform</quarkus.platform.group-id>
  <quarkus.platform.version>3.2.10.Final</quarkus.platform.version>
  <!-- 其他属性省略 -->
</properties>

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>${quarkus.platform.group-id}</groupId>
      <artifactId>${quarkus.platform.artifact-id}</artifactId>
      <version>${quarkus.platform.version}</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
    <dependency>
      <groupId>io.quarkus.platform</groupId>
      <artifactId>quarkus-camel-bom</artifactId>
      <version>${quarkus.platform.version}</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

<dependencies>
  <!-- Quarkus核心依赖 -->
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-arc</artifactId>
  </dependency>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-hibernate-orm</artifactId>
  </dependency>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-jdbc-postgresql</artifactId>
  </dependency>
  <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-liquibase</artifactId>
  </dependency>
  <!-- Camel Quarkus依赖 -->
  <dependency>
    <groupId>org.apache.camel.quarkus</groupId>
    <artifactId>camel-quarkus-bean</artifactId>
  </dependency>
  <dependency>
    <groupId>org.apache.camel.quarkus</groupId>
    <artifactId>camel-quarkus-ftp</artifactId>
  </dependency>
  <dependency>
    <groupId>org.apache.camel.quarkus</groupId>
    <artifactId>camel-quarkus-sql</artifactId>
  </dependency>
  <!-- 其他依赖省略 -->
</dependencies>

幂等仓库配置 

package com.epav.configuration;

import jakarta.inject.Named;
import jakarta.inject.Singleton;
import org.apache.camel.Configuration;
import org.apache.camel.processor.idempotent.jdbc.JdbcCachedMessageIdRepository;
import org.eclipse.microprofile.config.inject.ConfigProperty;

import javax.sql.DataSource;

@Configuration
public class FtpIdempotentRepositoryConfiguration {

    @ConfigProperty(name = "trader.integration.instance.name")
    String instanceName;

    @Singleton
    DataSource dataSource;

    @Singleton
    @Named(value = "FtpIdempotentRepository")
    public JdbcCachedMessageIdRepository jdbcCachedMessageRepository(){
        JdbcCachedMessageIdRepository repository = new JdbcCachedMessageIdRepository();
        repository.setDataSource(dataSource);
        repository.setProcessorName(instanceName);
        return repository;
    }
}

SFTP路由配置 

import com.intrasoft.configuration.properties.FtpRoutingProperties;
import jakarta.inject.Singleton;
import lombok.RequiredArgsConstructor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.idempotent.jdbc.JdbcCachedMessageIdRepository;

@Singleton
@RequiredArgsConstructor
public class IncomingRoute extends RouteBuilder {

    private final FtpRoutingProperties ftpRoutingProperties;
    private final JdbcCachedMessageIdRepository jdbcCachedMessageIdRepository;

    @Override
    public void configure() {
        String sftpUrl = String.format("sftp://%s:%s@%s:",
                ftpRoutingProperties.username,
                ftpRoutingProperties.password,
                ftpRoutingProperties.host);

        from(sftpUrl + ftpRoutingProperties.edbInputDir)
                .idempotentConsumer(header("CamelFileAbsolutePath"))
                .idempotentRepository(jdbcCachedMessageIdRepository)
                .completionEager(true)
                .process(CUSTOM_PROCESSOR_NAME)
                .to(sftpUrl + ftpRoutingProperties.successDir)
                .stop();
    }
}

应用配置 

quarkus.banner.enabled=false
quarkus.application.name=custom-app-name
trader.integration.instance.name=${pod.ip:${quarkus.application.name}}

quarkus.http.port=9090

quarkus.log.level=INFO
quarkus.log.console.format=%d{HH:mm:ss} %-5p [%c] (%t) %s%e%n

quarkus.devservices.enabled=false
%dev.quarkus.log.category."org.apache.camel".level=ERROR

%dev.quarkus.datasource.db-kind=postgresql
%dev.quarkus.datasource.username=idemp_db
%dev.quarkus.datasource.password=idemp_db
%dev.quarkus.datasource.jdbc.url=jdbc:postgresql://localhost:5432/idemp_db
%dev.quarkus.hibernate-orm.log.sql=false

quarkus.liquibase.migrate-at-start=true
quarkus.liquibase.change-log=db/master.xml

错误日志 

16:10:03 INFO  [liquibase.database] (Quarkus Main Thread) Set default schema name to public
16:10:03 INFO  [liquibase.executor] (Quarkus Main Thread) Changelog query completed.
16:10:03 INFO  [liquibase.executor] (Quarkus Main Thread) Changelog query completed.
16:10:03 INFO  [liquibase.executor] (Quarkus Main Thread) Changelog query completed.
16:10:03 INFO  [liquibase.lockservice] (Quarkus Main Thread) Successfully acquired change log lock
16:10:04 INFO  [liquibase.changelog] (Quarkus Main Thread) Reading from public.databasechangelog
16:10:04 INFO  [liquibase.executor] (Quarkus Main Thread) Changelog query completed.
Database is up to date, no changesets to execute
16:10:04 INFO  [liquibase.executor] (Quarkus Main Thread) Changelog query completed.
16:10:04 INFO  [liquibase.lockservice] (Quarkus Main Thread) Successfully released change log lock
16:10:07 ERROR [org.apache.camel.impl.engine.AbstractCamelContext] (Quarkus Main Thread) Error starting CamelContext (camel-1) due to exception thrown: Failed to start route sftp-incoming-edb-route because of java.lang.NullPointerException: Cannot invoke "org.springframework.transaction.support.TransactionTemplate.execute(org.springframework.transaction.support.TransactionCallback)" because "this.transactionTemplate" is null: org.apache.camel.FailedToStartRouteException: Failed to start route sftp-incoming-route because of java.lang.NullPointerException: Cannot invoke "org.springframework.transaction.support.TransactionTemplate.execute(org.springframework.transaction.support.TransactionCallback)" because "this.transactionTemplate" is null
    at org.apache.camel.impl.engine.RouteService.warmUp(RouteService.java:122)
    at org.apache.camel.impl.engine.InternalRouteStartupManager.doWarmUpRoutes(InternalRouteStartupManager.java:306)
    at org.apache.camel.impl.engine.InternalRouteStartupManager.safelyStartRouteServices(InternalRouteStartupManager.java:189)
    at org.apache.camel.impl.engine.InternalRouteStartupManager.doStartOrResumeRoutes(InternalRouteStartupManager.java:147)
    at org.apache.camel.impl.engine.AbstractCamelContext.doStartCamel(AbstractCamelContext.java:2827)
    at org.apache.camel.impl.engine.AbstractCamelContext.doStartContext(AbstractCamelContext.java:2509)
    at org.apache.camel.impl.engine.AbstractCamelContext.doStart(AbstractCamelContext.java:2464)
    at org.apache.camel.support.service.BaseService.start(BaseService.java:113)
    at org.apache.camel.impl.engine.AbstractCamelContext.start(AbstractCamelContext.java:2100)
    at org.apache.camel.impl.DefaultCamelContext.start(DefaultCamelContext.java:313)
    at org.apache.camel.quarkus.main.CamelMain.doStart(CamelMain.java:94)
    at org.apache.camel.support.service.BaseService.start(BaseService.java:113)
    at org.apache.camel.quarkus.main.CamelMain.startEngine(CamelMain.java:140)
    at org.apache.camel.quarkus.main.CamelMainRuntime.start(CamelMainRuntime.java:49)
    at org.apache.camel.quarkus.core.CamelBootstrapRecorder.start(CamelBootstrapRecorder.java:45)
    at io.quarkus.deployment.steps.CamelBootstrapProcessor$boot173480958.deploy_0(Unknown Source)
    at io.quarkus.deployment.steps.CamelBootstrapProcessor$boot173480958.deploy(Unknown Source)
    at io.quarkus.runner.ApplicationImpl.doStart(Unknown Source)
    at io.quarkus.runtime.Application.start(Application.java:101)
    at io.quarkus.runtime.ApplicationLifecycleManager.run(ApplicationLifecycleManager.java:111)
    at io.quarkus.runtime.Quarkus.run(Quarkus.java:71)
    at io.quarkus.runtime.Quarkus.run(Quarkus.java:44)
    at io.quarkus.runtime.Quarkus.run(Quarkus.java:124)
    at io.quarkus.runner.GeneratedMain.main(Unknown Source)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:568)
    at io.quarkus.runner.bootstrap.StartupActionImpl$1.run(StartupActionImpl.java:104)
    at java.base/java.lang.Thread.run(Thread.java:840)
解决方案

1. 替换幂等仓库实现（推荐）

JdbcCachedMessageIdRepository依赖Spring的TransactionTemplate，而Quarkus环境默认不提供该类，手动初始化时也未设置事务模板，导致NPE。建议改用Camel Quarkus原生支持的SqlIdempotentRepository：

步骤1：修改仓库配置类 

package com.epav.configuration;

import jakarta.inject.Named;
import jakarta.inject.Singleton;
import org.apache.camel.Configuration;
import org.apache.camel.component.sql.SqlIdempotentRepository;
import org.eclipse.microprofile.config.inject.ConfigProperty;

import javax.sql.DataSource;

@Configuration
public class FtpIdempotentRepositoryConfiguration {

    @ConfigProperty(name = "trader.integration.instance.name")
    String instanceName;

    private final DataSource dataSource;

    // 构造函数注入Quarkus自动提供的数据源
    public FtpIdempotentRepositoryConfiguration(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    @Singleton
    @Named(value = "FtpIdempotentRepository")
    public SqlIdempotentRepository sqlIdempotentRepository() {
        // 指定存储幂等记录的表名
        SqlIdempotentRepository repository = new SqlIdempotentRepository(dataSource, "ftp_idempotent_records");
        repository.setProcessorName(instanceName);
        // 设置本地缓存大小，减少数据库查询
        repository.setCacheSize(1000);
        return repository;
    }
}

步骤2：更新路由注入 

// 替换原JdbcCachedMessageIdRepository为SqlIdempotentRepository
private final SqlIdempotentRepository sqlIdempotentRepository;

@Override
public void configure() {
    String sftpUrl = String.format("sftp://%s:%s@%s:",
            ftpRoutingProperties.username,
            ftpRoutingProperties.password,
            ftpRoutingProperties.host);

    from(sftpUrl + ftpRoutingProperties.edbInputDir)
            .idempotentConsumer(header("CamelFileAbsolutePath"))
            .idempotentRepository(sqlIdempotentRepository)
            .completionEager(true)
            .process(CUSTOM_PROCESSOR_NAME)
            .to(sftpUrl + ftpRoutingProperties.successDir)
            .stop();
}

步骤3：创建数据库表

在Liquibase的db/master.xml中添加表结构：

<changeSet id="create-ftp-idempotent-table" author="your-team">
    <createTable tableName="ftp_idempotent_records">
        <column name="processor_name" type="VARCHAR(255)" nullable="false"/>
        <column name="message_id" type="VARCHAR(255)" nullable="false"/>
        <column name="created_at" type="TIMESTAMP" defaultValueComputed="CURRENT_TIMESTAMP"/>
        <constraint primaryKeyName="pk_ftp_idempotent" primaryKey="processor_name, message_id"/>
    </createTable>
</changeSet>

2. 修复DataSource注入问题

原配置中DataSource字段仅标注@Singleton，未通过@Inject或构造函数注入，Quarkus无法自动注入实例。即使使用原JdbcCachedMessageIdRepository，也需要修正注入方式：

@Inject
@Singleton
DataSource dataSource;
无需额外服务的分布式锁实现建议
  • 数据库唯一约束锁：
    利用数据库唯一键约束实现分布式锁，适合中小规模集群。可以封装通用工具类，通过JDBC执行INSERT ... ON CONFLICT语句抢占锁，配合超时清理机制避免死锁。
  • 本地锁+DB协调：
    先获取JVM本地锁（如ReentrantLock），再抢占数据库锁，减少数据库竞争压力。适合单实例高并发、多实例低竞争的场景。
  • 嵌入式Infinispan锁：
    使用camel-quarkus-infinispan依赖，启用Infinispan嵌入式模式（无需额外服务），基于分布式缓存实现高性能锁：
    1. 添加依赖：
    <dependency>
    <groupId>org.apache.camel.quarkus</groupId>
    <artifactId>camel-quarkus-infinispan</artifactId>
</dependency>
    1. 配置嵌入式缓存：
    quarkus.infinispan.embedded.enabled=true
quarkus.infinispan.embedded.config=infinispan.xml
    1. 创建Infinispan幂等仓库：
    @Singleton
@Named("InfinispanIdempotentRepository")
public InfinispanIdempotentRepository infinispanIdempotentRepository(EmbeddedCacheManager cacheManager) {
    Cache<String, String> cache = cacheManager.getCache("ftp-lock-cache");
    return new InfinispanIdempotentRepository(cache);
}

内容的提问来源于stack exchange，提问作者JavaEnthusiast27

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
火山引擎 最新活动
全面开放
Doubao-Seedance-2.5全新模型升级

模型再升级,30秒超长叙事, 模态参考扩容

HOT
方舟Codingplan

模型自由，工具不限，最新支持 Deepseek-V4 系列、GLM-5.3 系列

HOT
方舟Agent plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列

HOT
新用户特惠专场

大模型19元起，Al应用9.9元畅享，新人首购爆款尽享优惠

最近更新时间：2026.06.27 22:25:56