如何用Spring实现Zeebe Exporter将JobRecords存入PostgreSQL

1. 添加依赖 #

在Maven/Gradle项目中引入必要依赖（版本匹配你使用的Zeebe 8.x和Spring 6.x）：
Maven示例：

<!-- Zeebe Exporter API -->
<dependency>
    <groupId>io.camunda</groupId>
    <artifactId>zeebe-exporter-api</artifactId>
    <version>8.x.x</version>
    <scope>provided</scope>
</dependency>

<!-- Spring Context -->
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context</artifactId>
    <version>6.0.x</version>
</dependency>

<!-- Spring Data JPA -->
<dependency>
    <groupId>org.springframework.data</groupId>
    <artifactId>spring-data-jpa</artifactId>
    <version>3.0.x</version>
</dependency>

<!-- PostgreSQL驱动 -->
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.6.x</version>
</dependency>

2. 定义数据库实体 #

创建对应PostgreSQL表的JPA实体，存储UserTask的核心信息：

import jakarta.persistence.*;
import java.time.Instant;

@Entity
@Table(name = "user_task_records")
public class UserTaskRecord {
    @Id
    @Column(name = "job_key")
    private long jobKey;

    @Column(name = "task_type")
    private String taskType;

    @Column(name = "process_instance_key")
    private long processInstanceKey;

    @Column(name = "created_at")
    private Instant createdAt;

    // 省略getter、setter及构造方法
}

3. 实现数据访问层 #

用Spring Data JPA Repository简化数据库操作：

import org.springframework.data.jpa.repository.JpaRepository;

public interface UserTaskRecordRepository extends JpaRepository<UserTaskRecord, Long> {
}

4. 初始化Spring上下文 #

由于Zeebe Exporter由Broker加载，需手动初始化Spring上下文以获取Repository实例：

import org.springframework.context.annotation.AnnotationConfigApplicationContext;

public class SpringContextProvider {
    private static final AnnotationConfigApplicationContext context;

    static {
        context = new AnnotationConfigApplicationContext();
        context.scan("com.your.package"); // 替换为实体、Repository所在的包路径
        context.refresh();
    }

    public static <T> T getBean(Class<T> beanClass) {
        return context.getBean(beanClass);
    }
}

5. 实现Zeebe Exporter类 #

继承AbstractExporter，监听Job的CREATED事件，提取UserTask信息并入库：

import io.camunda.zeebe.exporter.api.AbstractExporter;
import io.camunda.zeebe.exporter.api.context.Context;
import io.camunda.zeebe.exporter.api.context.Controller;
import io.camunda.zeebe.exporter.api.record.Record;
import io.camunda.zeebe.exporter.api.record.RecordType;
import io.camunda.zeebe.exporter.api.record.value.JobRecordValue;

public class UserTaskExporter extends AbstractExporter {
    private UserTaskRecordRepository repository;

    @Override
    public void configure(Context context) {
        repository = SpringContextProvider.getBean(UserTaskRecordRepository.class);
    }

    @Override
    public void open(Controller controller) {
        super.open(controller);
        // 设置从最新记录开始导出，可根据需求调整
        controller.updateLastExportedRecordPosition(controller.getLastExportedRecordPosition());
    }

    @Override
    public void export(Record<?> record) {
        // 仅处理Job类型的CREATED事件，且过滤UserTask类型的任务
        if (record.getRecordType() == RecordType.EVENT &&
            record.getValue() instanceof JobRecordValue jobRecord) {
            if ("userTask".equals(jobRecord.getType())) {
                UserTaskRecord taskRecord = new UserTaskRecord();
                taskRecord.setJobKey(record.getKey());
                taskRecord.setTaskType(jobRecord.getType());
                taskRecord.setProcessInstanceKey(jobRecord.getProcessInstanceKey());
                taskRecord.setCreatedAt(record.getTimestamp());

                repository.save(taskRecord);
            }
        }
    }
}

6. 配置Zeebe Broker加载Exporter #

在Zeebe Broker的application.yaml中添加Exporter配置：

zeebe:
  brokers:
    exporters:
      userTaskExporter:
        className: com.your.package.UserTaskExporter
        args:
          # 可添加自定义配置参数，如数据库连接信息（按需）

7. 关键注意事项 #

• 确保Spring扫描包路径正确，否则无法获取Repository Bean。
• 高并发场景下建议批量插入（缓存一定数量记录后批量保存），避免频繁数据库交互。
• 如需事务支持，可在Repository方法上添加@Transactional注解。
• 严格保证Zeebe Exporter API版本与Broker版本一致，避免兼容性问题。

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