Welcome to MikeDaDev's Blog – notes on Java, backend & system design.

Explore
MikeDaDev's Blog
in java

activiti 8

Tung Bui Thanh (TungDaDev)22 mins read

Activiti.webp

# activiti là gì và tại sao cần nó?

Workflow nào thì cũng code được bằng if-else, state machine hay event-driven. Nhưng khi business process dài 20+ bước, có rẽ nhánh, chạy song song, chờ người duyệt, timeout, xử lý lỗi, mà lại hay phải đổi và đổi xong KHÔNG muốn redeploy code, thì lúc đó nên nghĩ tới một process engine.

Activiti là BPMN 2.0 engine chạy trên JVM. Nó nhận file BPMN XML (vẽ bằng visual designer), parse thành executable logic rồi điều phối execution trong suốt process lifecycle. Dev không phải code phần điều hướng luồng, chỉ code từng task riêng lẻ (delegate). Còn luồng thì nằm trong BPMN diagram, BA muốn sửa gì thì sửa ở đó.

Ví dụ: một ngân hàng dùng Activiti cho quy trình duyệt khoản vay. Khách submit đơn trên web portal là BPMN XML chạy: tự gọi credit scoring API, chuyển hồ sơ tới đúng nhân viên duyệt, gửi notification và theo dõi SLA.

# activiti vs camunda vs flowable

Cả 3 đều chung gốc. Activiti do chính nhóm tác giả jBPM (Tom Baeyens, Joram Barrez) viết lại từ đầu năm 2010. Camunda fork từ Activiti 5 năm 2013, Flowable fork từ Activiti 6 năm 2016. Riêng Camunda 8 thì đổi hẳn sang engine mới là Zeebe, không còn chung code. Khác nhau ở mấy điểm sau:

Activiti 8Camunda 7/8Flowable 7
LicenseApache 2.0Community + EnterpriseApache 2.0
Cloud-nativeActiviti Cloud (K8s)Camunda 8 (Zeebe)Flowable Cloud
Spring BootNative integrationGood integrationNative integration
Runtime modelShared DB (RDBMS)Shared DB / Event-sourced (C8)Shared DB
Ghi chúBản cloud của AlfrescoC7 CE hết hỗ trợ từ 10/2025; C8 tự host cần license cho productionFork từ Activiti 6

# architecture - activiti bên trong làm gì?

Khi gọi runtimeService.startProcessInstanceByKey("loan-approval"), bên dưới diễn ra thế này:

1. Engine đọc Process Definition từ ACT_RE_PROCDEF table
2. Tạo Process Instance (row trong ACT_RU_EXECUTION)
3. Tạo execution token ở Start Event
4. Token di chuyển theo Sequence Flow
5. Gặp Service Task → gọi Java Delegate → execute business logic
6. Gặp User Task → INSERT vào ACT_RU_TASK → DỪNG (wait state)
7. Khi user complete task → token tiếp tục di chuyển
8. Gặp Gateway → evaluate conditions → chọn paths
9. Gặp End Event → complete instance → move to history tables (ACT_HI_*)

# database tables

Activiti 8 tạo khoảng 25 table (bản 5/6 khoảng 28), chia theo prefix:

PrefixTênMục đíchVí dụ
ACT_RE_RepositoryProcess definitions, deploymentsACT_RE_PROCDEF, ACT_RE_DEPLOYMENT
ACT_RU_RuntimeRunning instances, tasks, variables, jobsACT_RU_EXECUTION, ACT_RU_TASK
ACT_HI_HistoryCompleted instances, audit trailACT_HI_PROCINST, ACT_HI_ACTINST
ACT_GE_GeneralByte arrays, propertiesACT_GE_BYTEARRAY
ACT_ID_IdentityUsers, groups (chỉ có ở Activiti 5/6, từ bản 7 đã bỏ, user/group lấy từ Spring Security)ACT_ID_USER, ACT_ID_GROUP

Chỗ này ảnh hưởng performance: các table ACT_RU_* chỉ chứa data đang chạy, instance xong là data chuyển qua ACT_HI_*. Có 1 triệu instance đã xong thì ACT_RU_EXECUTION vẫn nhỏ (chỉ còn instance đang chạy). Nhưng ACT_HI_* thì sẽ phình to, phải tính chuyện archive.

# tích hợp spring boot: setup

# dependencies

<dependency>
   <groupId>org.activiti</groupId>
   <artifactId>activiti-spring-boot-starter</artifactId>
   <version>8.6.0</version>
</dependency>
 
<!-- Activiti 7.1.0.M6 trở về sau KHÔNG có trên Maven Central, phải khai báo repo của Alfresco -->
<repositories>
   <repository>
      <id>activiti-releases</id>
      <url>https://artifacts.alfresco.com/nexus/content/repositories/activiti-releases</url>
   </repository>
</repositories>

Lưu ý chỗ repository: bản cuối cùng của Activiti có trên Maven Central là 7.1.0.M6. Các bản 8.x (mới nhất lúc mình viết là 8.8.1) chỉ nằm trên Nexus của Alfresco, quên khai báo repo là Maven báo không tìm thấy artifact ngay.

# configuration

spring:
  activiti:
    database-schema-update: true # Auto create/update schema
    history-level: full # Lưu full audit (activity, variable, task details)
    async-executor-activate: true # Enable async job execution
    check-process-definitions: false # Không auto-deploy từ classpath
 
  datasource:
    url: jdbc:postgresql://localhost:5432/bpm_db
    username: ${DB_USERNAME}
    password: ${DB_PASSWORD}
    hikari:
      maximum-pool-size: 30 # Activiti cần nhiều connections cho async jobs

# configuration class

Starter đã tự tạo sẵn bean SpringProcessEngineConfiguration (gắn DataSource, transaction manager, tự deploy file BPMN trong processes/...). Muốn chỉnh thêm thì đừng tự new SpringProcessEngineConfiguration() rồi khai báo @Bean: bean của starter có @ConditionalOnMissingBean, bạn khai báo bean riêng là nó lùi lại, mất luôn mấy thứ starter đã cấu hình sẵn. Cách đúng là khai báo một ProcessEngineConfigurationConfigurer, starter sẽ gọi nó trên chính configuration của mình:

@Configuration
public class ActivitiConfig {
 
   @Bean
   public ProcessEngineConfigurationConfigurer activitiCustomizer() {
       return config -> {
           // DataSource, transaction manager, schema update, history level
           // đã lấy từ application.yml, không cần set lại ở đây
 
           // Custom ID generator (UUID thay vì sequential)
           // Hoặc chỉ cần spring.activiti.use-strong-uuids: true
           config.setIdGenerator(new StrongUuidGenerator());
 
           // Poll interval của async executor (ms)
           config.setAsyncExecutorDefaultAsyncJobAcquireWaitTime(5000);
       };
   }
}

Activiti tự nhận Spring bean và cho gọi thẳng trong BPMN expression ${beanName.method()}. Không phải register tay gì cả, Spring ApplicationContext chính là bean registry của Activiti.

# core services: mấy api chính cần biết

Activiti 5/6 có 7 service. Từ Activiti 7, IdentityService và FormService bị bỏ, ProcessEngine còn 6 service: RepositoryService, RuntimeService, TaskService, HistoryService, ManagementService và DynamicBpmnService. Làm hằng ngày thì chủ yếu đụng 4-5 cái đầu.

Nói thêm cho đủ: Activiti 7+ còn có lớp API mới ProcessRuntime / TaskRuntime (gói org.activiti.api), gắn với Spring Security và tự kiểm tra quyền theo user đang đăng nhập. Bài này dùng các service "cổ điển" của engine vì chúng vẫn chạy tốt, linh hoạt hơn, và cũng là thứ bạn gặp khi đọc code Activiti 6 hay Flowable.

# repositoryservice: quản lý process definition

Cứ hiểu đây là chỗ lo deploy và versioning. Mỗi file BPMN deploy lên thành 1 ProcessDefinition. Deploy lại thì vẫn cùng key, chỉ tăng version.

@Service
@RequiredArgsConstructor
@Slf4j
public class ProcessDefinitionService {
 
   private final RepositoryService repositoryService;
 
   /**
    * Deploy BPMN XML. Mỗi lần deploy cùng key → version tăng.
    * Instances đang chạy KHÔNG bị ảnh hưởng — giữ version cũ.
    * Instances mới start sẽ dùng version mới nhất.
    */
   public DeploymentResult deploy(String bpmnXml, String processName, String workspaceId) {
       Deployment deployment = repositoryService.createDeployment()
           .name(processName)
           .tenantId(workspaceId)  // Multi-tenant isolation
           .addString(processName + ".bpmn20.xml", bpmnXml)
           .deploy();
 
       // Lấy definition mới deploy
       ProcessDefinition definition = repositoryService.createProcessDefinitionQuery()
           .deploymentId(deployment.getId())
           .singleResult();
 
       log.info("Deployed process | name={} | version={} | deploymentId={}",
           definition.getName(), definition.getVersion(), deployment.getId());
 
       return new DeploymentResult(deployment.getId(), definition.getId(), definition.getVersion());
   }
 
   /**
    * Lấy tất cả versions của 1 process definition.
    * Hữu ích cho rollback: activate version cũ, suspend version mới.
    */
   public List<ProcessDefinitionDTO> getVersions(String processKey, String workspaceId) {
       return repositoryService.createProcessDefinitionQuery()
           .processDefinitionKey(processKey)
           .processDefinitionTenantId(workspaceId)
           .orderByProcessDefinitionVersion().desc()
           .list()
           .stream()
           .map(this::toDTO)
           .toList();
   }
 
   /**
    * Suspend definition → không thể start instances mới.
    * Useful khi phát hiện bug trong process, cần "pause" nó.
    */
   public void suspend(String definitionId) {
       repositoryService.suspendProcessDefinitionById(definitionId, true, null);
       // suspendProcessInstances=true → suspend cả running instances
       log.info("Suspended process definition: {}", definitionId);
   }
 
   /**
    * Lấy BPMN XML từ deployed definition.
    * Useful cho: hiển thị diagram, audit, export.
    */
   public String getBpmnXml(String definitionId) {
       ProcessDefinition def = repositoryService.getProcessDefinition(definitionId);
       try (InputStream is = repositoryService.getResourceAsStream(
               def.getDeploymentId(), def.getResourceName())) {
           return new String(is.readAllBytes(), StandardCharsets.UTF_8);
       } catch (IOException e) {
           throw new RuntimeException("Failed to read BPMN resource", e);
       }
   }
}

# runtimeservice: start và điều khiển instance

Service này dùng nhiều nhất: start process, bắn signal/event, get/set variable.

@Service
@RequiredArgsConstructor
@Slf4j
public class ProcessInstanceService {
 
   private final RuntimeService runtimeService;
 
   /**
    * Start process instance.
    * businessKey = unique identifier từ business domain (order ID, loan ID).
    * Variables = initial data cho process.
    */
   public ProcessInstanceDTO startProcess(StartProcessRequest request) {
       Map<String, Object> variables = new HashMap<>();
       variables.put("applicantName", request.getApplicantName());
       variables.put("loanAmount", request.getLoanAmount());
       variables.put("applicationDate", LocalDate.now().toString());
       variables.put("workspaceId", request.getWorkspaceId());
 
       ProcessInstance instance = runtimeService.startProcessInstanceByKey(
           request.getProcessKey(),       // BPMN process ID (VD: "loan-approval")
           request.getBusinessKey(),      // Business identifier (VD: "LOAN-2024-001")
           variables
       );
 
       log.info("Process started | key={} | instanceId={} | businessKey={}",
           request.getProcessKey(), instance.getId(), request.getBusinessKey());
 
       return toDTO(instance);
   }
 
   /**
    * Gửi signal/message đến running instance.
    * Use case: external system callback, webhook response.
    */
   public void sendMessage(String messageName, String businessKey, Map<String, Object> payload) {
       // Find execution đang wait message
       Execution execution = runtimeService.createExecutionQuery()
           .messageEventSubscriptionName(messageName)
           .processInstanceBusinessKey(businessKey)
           .singleResult();
 
       if (execution == null) {
           throw new ProcessNotFoundException(
               "No process waiting for message: " + messageName + " with key: " + businessKey);
       }
 
       runtimeService.messageEventReceived(messageName, execution.getId(), payload);
       log.info("Message delivered | message={} | businessKey={}", messageName, businessKey);
   }
 
   /**
    * Get variables từ running instance.
    * Variables = shared state giữa tasks trong cùng instance.
    */
   public Map<String, Object> getVariables(String instanceId) {
       return runtimeService.getVariables(instanceId);
   }
 
   /**
    * Set/update variable dynamically.
    * Use case: external event update data giữa chừng process.
    */
   public void setVariable(String instanceId, String varName, Object value) {
       runtimeService.setVariable(instanceId, varName, value);
   }
 
   /**
    * Cancel running instance.
    * Use case: customer cancel application, admin abort stuck process.
    */
   public void cancel(String instanceId, String reason) {
       runtimeService.deleteProcessInstance(instanceId, reason);
       log.info("Process cancelled | instanceId={} | reason={}", instanceId, reason);
   }
}

# taskservice: user task (chỗ cần người thao tác)

User Task là "điểm dừng": process đứng lại chờ người vào xử lý.

TaskService lo mấy việc: query task cho user, claim/release, complete kèm output variable.

Có một khái niệm cần nắm: Assignee vs Candidate. Assignee là task giao đích danh cho 1 người. Candidate Groups là task để chung cho cả nhóm, ai claim trước thì người đó làm. Giống email riêng với hộp thư chung vậy.

@Service
@RequiredArgsConstructor
@Slf4j
public class UserTaskService {
 
   private final TaskService taskService;
 
   /**
    * Lấy tasks assigned cho user cụ thể.
    * "Inbox" của user — tasks đang chờ user hoàn thành.
    */
   public Page<TaskDTO> getMyTasks(String userId, Pageable pageable) {
       long total = taskService.createTaskQuery()
           .taskAssignee(userId)
           .count();
 
       List<Task> tasks = taskService.createTaskQuery()
           .taskAssignee(userId)
           .orderByTaskCreateTime().desc()
           .listPage((int) pageable.getOffset(), pageable.getPageSize());
 
       List<TaskDTO> dtos = tasks.stream().map(this::toDTO).toList();
       return new PageImpl<>(dtos, pageable, total);
   }
 
   /**
    * Lấy tasks available cho group (chưa ai claim).
    * "Shared queue" — ai rảnh thì nhận.
    */
   public List<TaskDTO> getGroupTasks(String groupId) {
       return taskService.createTaskQuery()
           .taskCandidateGroup(groupId)
           .taskUnassigned()  // Chưa ai claim
           .orderByTaskCreateTime().desc()
           .list()
           .stream()
           .map(this::toDTO)
           .toList();
   }
 
   /**
    * Claim task từ group queue.
    * Sau khi claim, task không còn visible cho người khác trong group.
    */
   public void claim(String taskId, String userId) {
       taskService.claim(taskId, userId);
       log.info("Task claimed | taskId={} | userId={}", taskId, userId);
   }
 
   /**
    * Unclaim — trả task về queue (user bận, không thể handle).
    */
   public void unclaim(String taskId) {
       taskService.unclaim(taskId);
   }
 
   /**
    * Complete task với output variables.
    * Variables này available cho downstream tasks/gateways.
    *
    * Ví dụ: Approver complete với decision="approve", comments="Looks good"
    * → Gateway đọc ${decision == 'approve'} → route accordingly
    */
   public void complete(String taskId, Map<String, Object> variables, String userId) {
       // Verify task assigned to this user
       Task task = taskService.createTaskQuery()
           .taskId(taskId)
           .taskAssignee(userId)
           .singleResult();
 
       if (task == null) {
           throw new TaskNotFoundException("Task not found or not assigned to you: " + taskId);
       }
 
       taskService.complete(taskId, variables);
       log.info("Task completed | taskId={} | taskName={} | userId={} | vars={}",
           taskId, task.getName(), userId, variables.keySet());
   }
 
   /**
    * Delegate task cho người khác.
    * Use case: manager delegate cho subordinate.
    */
   public void delegate(String taskId, String fromUserId, String toUserId) {
       taskService.delegateTask(taskId, toUserId);
       log.info("Task delegated | taskId={} | from={} | to={}", taskId, fromUserId, toUserId);
   }
 
   /**
    * Add comment to task (audit trail).
    */
   public void addComment(String taskId, String userId, String message) {
       Task task = taskService.createTaskQuery().taskId(taskId).singleResult();
       taskService.addComment(taskId, task.getProcessInstanceId(), message);
   }
 
   private TaskDTO toDTO(Task task) {
       Map<String, Object> variables = taskService.getVariables(task.getId());
       return TaskDTO.builder()
           .id(task.getId())
           .name(task.getName())
           .assignee(task.getAssignee())
           .createTime(task.getCreateTime().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime())
           .dueDate(task.getDueDate() != null ? task.getDueDate().toInstant().atZone(ZoneId.systemDefault()).toLocalDate() : null)
           .processInstanceId(task.getProcessInstanceId())
           .processDefinitionId(task.getProcessDefinitionId())
           .formKey(task.getFormKey())
           .variables(variables)
           .build();
   }
}

# historyservice: audit trail và analytics

Chuyện gì đã xảy ra cũng nằm trong history. Process xong thì runtime table được dọn, còn history vẫn giữ. Đây là nguồn data cho audit, phân tích performance, tìm chỗ nghẽn.

@Service
@RequiredArgsConstructor
public class ProcessHistoryService {
 
   private final HistoryService historyService;
 
   /**
    * Lấy toàn bộ history của 1 process instance.
    * Ai làm gì, khi nào, mất bao lâu — full timeline.
    */
   public ProcessHistoryDTO getInstanceHistory(String instanceId) {
       HistoricProcessInstance instance = historyService.createHistoricProcessInstanceQuery()
           .processInstanceId(instanceId)
           .singleResult();
 
       List<HistoricActivityInstance> activities = historyService.createHistoricActivityInstanceQuery()
           .processInstanceId(instanceId)
           .orderByHistoricActivityInstanceStartTime().asc()
           .list();
 
       List<HistoricTaskInstance> tasks = historyService.createHistoricTaskInstanceQuery()
           .processInstanceId(instanceId)
           .orderByHistoricTaskInstanceStartTime().asc()
           .list();
 
       return ProcessHistoryDTO.builder()
           .instanceId(instanceId)
           .startTime(instance.getStartTime())
           .endTime(instance.getEndTime())
           .durationMs(instance.getDurationInMillis())
           .status(instance.getEndTime() != null ? "COMPLETED" : "RUNNING")
           .activities(activities.stream().map(this::toActivityDTO).toList())
           .tasks(tasks.stream().map(this::toTaskDTO).toList())
           .build();
   }
 
   /**
    * Phân tích performance: average duration per task.
    * Phát hiện bottleneck: task nào mất thời gian nhiều nhất?
    */
   public List<TaskPerformanceDTO> getTaskPerformance(String processKey, LocalDate from, LocalDate to) {
       return historyService.createHistoricTaskInstanceQuery()
           .processDefinitionKey(processKey)
           .taskCompletedAfter(toDate(from))
           .taskCompletedBefore(toDate(to))
           .list()
           .stream()
           .collect(Collectors.groupingBy(HistoricTaskInstance::getName))
           .entrySet().stream()
           .map(entry -> {
               List<HistoricTaskInstance> tasks = entry.getValue();
               double avgDuration = tasks.stream()
                   .mapToLong(t -> t.getDurationInMillis() != null ? t.getDurationInMillis() : 0)
                   .average().orElse(0);
               return new TaskPerformanceDTO(
                   entry.getKey(),
                   tasks.size(),
                   (long) avgDuration,
                   tasks.stream().mapToLong(HistoricTaskInstance::getDurationInMillis).max().orElse(0)
               );
           })
           .sorted(Comparator.comparingLong(TaskPerformanceDTO::avgDurationMs).reversed())
           .toList();
   }
}

# java delegates: chỗ đặt business logic

Java Delegate là cầu nối giữa BPMN diagram và Java code. Mỗi Service Task trong diagram gọi 1 delegate. Delegate nhận execution context (variable, thông tin process), chạy logic rồi set output variable.

vài nguyên tắc khi viết delegate:

  1. Single Responsibility: 1 delegate làm đúng một việc. Đừng nhồi 500 dòng logic vào 1 delegate.
  2. Idempotent: engine có thể retry, nên chargePayment() phải check "charge chưa?" rồi mới charge.
  3. Fast: delegate chạy sync sẽ block thread đang gọi engine (và giữ luôn transaction lẫn DB connection). Gọi external API mà mất 30s thì chuyển qua async task.
  4. Stateless: delegate là Spring singleton, đừng giữ state giữa các lần gọi. Cần lưu gì thì dùng process variable.
/**
* Delegate tính toán credit score.
*
* Input variables: customerId, loanAmount
* Output variables: creditScore, riskLevel, autoApproved
*
* BPMN reference: activiti:delegateExpression="${creditScoreDelegate}"
*/
@Component("creditScoreDelegate")
@RequiredArgsConstructor
@Slf4j
public class CreditScoreDelegate implements JavaDelegate {
 
   private final CreditScoringClient scoringClient;  // Gọi external service
   private final MetricsService metrics;
 
   @Override
   public void execute(DelegateExecution execution) {
       // 1. Read input variables
       String customerId = (String) execution.getVariable("customerId");
       BigDecimal loanAmount = (BigDecimal) execution.getVariable("loanAmount");
       String processId = execution.getProcessInstanceId();
 
       log.info("[processId={}] Calculating credit score | customerId={} | amount={}",
           processId, customerId, loanAmount);
 
       // 2. Execute business logic
       long start = System.nanoTime();
       try {
           CreditScoreResult result = scoringClient.calculate(customerId, loanAmount);
 
           // 3. Set output variables (available for downstream tasks/gateways)
           execution.setVariable("creditScore", result.getScore());
           execution.setVariable("riskLevel", result.getRiskLevel().name());
           execution.setVariable("autoApproved", result.getScore() >= 750);
           execution.setVariable("scoringDetails", result.getDetails());
 
           long durationMs = (System.nanoTime() - start) / 1_000_000;
           metrics.recordTaskDuration("credit_score_calculation", durationMs);
 
           log.info("[processId={}] Credit score calculated | score={} | risk={} | autoApproved={} | {}ms",
               processId, result.getScore(), result.getRiskLevel(),
               result.getScore() >= 750, durationMs);
 
       } catch (ScoringServiceUnavailableException e) {
           // Throw BpmnError → caught by boundary error event trên diagram
           log.error("[processId={}] Scoring service unavailable: {}", processId, e.getMessage());
           throw new BpmnError("SCORING_UNAVAILABLE",
               "Credit scoring service is temporarily unavailable");
 
       } catch (InvalidCustomerException e) {
           // Business error → route to rejection path
           execution.setVariable("rejectionReason", e.getMessage());
           throw new BpmnError("INVALID_CUSTOMER", e.getMessage());
       }
       // RuntimeException (NPE, etc.) → không phải BpmnError nên không đi theo boundary event.
       // Task sync: transaction rollback về wait state trước đó, exception ném ngược ra cho caller.
       // Task async: job bị trừ retries, hết retries thì vào dead letter (ACT_RU_DEADLETTER_JOB).
   }
}

# delegate expression vs class vs expression

Có 3 cách trỏ tới delegate trong BPMN XML:

<!-- 1. delegateExpression: Spring bean name (RECOMMENDED) -->
<!-- Bean phải implement JavaDelegate -->
<serviceTask activiti:delegateExpression="${creditScoreDelegate}"/>
 
<!-- 2. class: Full qualified class name (instantiated by engine, NOT Spring-managed) -->
<!-- KHÔNG inject Spring beans được (no @Autowired) -->
<serviceTask activiti:class="com.example.bpm.delegate.CreditScoreDelegate"/>
 
<!-- 3. expression: Call method trên Spring bean (không cần implement JavaDelegate) -->
<!-- Return value tự động set vào biến nếu có resultVariable -->
<serviceTask activiti:expression="${orderService.calculateTotal(execution)}"
            activiti:resultVariable="totalAmount"/>

Khuyên dùng: cứ delegateExpression mà xài, vì nó do Spring quản lý, inject được, test được.

# async execution & job executor

Mặc định Service Task chạy synchronous: Activiti không có thread riêng cho phần này, delegate chạy luôn trên thread đã gọi startProcessInstanceByKey() hay taskService.complete(), thường là thread xử lý HTTP request, và chạy trong cùng một transaction. Delegate gọi external API mất 30 giây thì request đó đứng 30 giây, giữ luôn một DB connection suốt 30 giây. Vài chục request như vậy cùng lúc là cạn thread pool của Tomcat hoặc connection pool của Hikari, cả app đứng hình chứ không riêng gì engine.

Async task gỡ chỗ này: engine tạo một "job" trong database rồi return luôn. Job Executor (thread pool chạy nền) nhặt job ra chạy async. Fail thì job được retry, mặc định 3 lần, mỗi lần cách nhau một khoảng cố định (không phải exponential backoff; muốn đổi lịch retry thì dùng activiti:failedJobRetryTimeCycle, vd R5/PT10M là 5 lần cách nhau 10 phút). Hết lượt retry thì job bị chuyển vào bảng dead letter (ACT_RU_DEADLETTER_JOB).

Có điều async không làm external call nhanh hơn. Nó chỉ chuyển chỗ chờ từ thread của request sang thread của job executor, mà pool này cũng có giới hạn. Lợi ích thật là request trả về ngay, transaction ngắn lại, và lỗi thì có cơ chế retry thay vì ném thẳng vào mặt người dùng.

<!-- Async service task: engine không block, delegate chạy bởi job executor -->
<serviceTask id="callExternalApi"
            activiti:delegateExpression="${externalApiDelegate}"
            activiti:async="true"
            activiti:exclusive="true"/>
<!--
 async="true": tạo job thay vì execute inline
 exclusive="true": không chạy song song với jobs cùng instance (tránh optimistic lock).
                   Đây cũng là giá trị mặc định, ghi ra cho rõ.
-->

# khi nào dùng async?

ScenarioSyncAsyncLý do
Tính toán nhanh (<100ms)✓Overhead job creation không đáng
Gọi external API✓Network latency unpredictable
Gửi email/notification✓Không cần block process flow
Database query nội bộ✓Nhanh, cùng transaction
File processing (large)✓CPU-intensive, có thể OOM
Sau parallel gateway fork✓Mỗi path chạy independent job (*)

(*) Job async mặc định là exclusive, nên các nhánh của cùng một instance vẫn chạy lần lượt, không song song thật. Muốn song song thật thì đặt activiti:exclusive="false" và chấp nhận optimistic lock ở chỗ join (engine sẽ retry job bị đụng).

# job executor configuration

Với Spring Boot starter, job async chạy trên TaskExecutor của Spring (bean applicationTaskExecutor), không phải thread pool riêng của engine. Vì vậy số thread chỉnh qua spring.task.execution.pool.core-size / max-size, còn setAsyncExecutorCorePoolSize() / setAsyncExecutorMaxPoolSize() chỉ có tác dụng khi chạy engine thuần, không qua starter. Phần còn lại (poll, retry, lock) thì vẫn chỉnh như sau:

@Configuration
public class AsyncExecutorConfig {
 
   @Bean
   public ProcessEngineConfigurationConfigurer asyncExecutorCustomizer() {
       return config -> {
           // Async executor settings
           config.setAsyncExecutorActivate(true);
           config.setAsyncExecutorDefaultAsyncJobAcquireWaitTime(5000);  // Poll interval async job (ms)
           config.setAsyncExecutorDefaultTimerJobAcquireWaitTime(5000);  // Poll interval timer job (ms)
           config.setAsyncExecutorMaxAsyncJobsDuePerAcquisition(5);      // Jobs per batch
 
           // Retry configuration
           config.setAsyncFailedJobWaitTime(10);     // Async job fail: chờ 10 giây rồi retry
           config.setDefaultFailedJobWaitTime(300);  // Timer job fail: chờ 5 phút rồi retry
           config.setAsyncExecutorResetExpiredJobsInterval(60000);       // 1 minute: nhả lock của job "treo"
       };
   }
}

# handling failed jobs

@Service
@RequiredArgsConstructor
@Slf4j
public class FailedJobService {
 
   private final ManagementService managementService;
 
   /**
    * Lấy jobs đang stuck (retries exhausted).
    * Monitor dashboard nên alert khi có deadletter jobs.
    */
   public List<DeadLetterJobDTO> getDeadLetterJobs() {
       // Job của Activiti 8 không có createTime, nên sắp theo duedate
       return managementService.createDeadLetterJobQuery()
           .orderByJobDuedate().desc()
           .list()
           .stream()
           .map(job -> DeadLetterJobDTO.builder()
               .jobId(job.getId())
               .processInstanceId(job.getProcessInstanceId())
               .exceptionMessage(job.getExceptionMessage())
               .duedate(job.getDuedate())
               .retries(job.getRetries())
               .build())
           .toList();
   }
 
   /**
    * Retry dead letter job (sau khi fix root cause).
    * Move job từ deadletter → executable queue.
    */
   public void retryDeadLetterJob(String jobId) {
       managementService.moveDeadLetterJobToExecutableJob(jobId, 3); // 3 retries
       log.info("Dead letter job moved to executable | jobId={}", jobId);
   }
}

# multi-tenancy - workspace isolation

Làm SaaS thì mỗi organization/tenant phải tách biệt nhau. Process definition và instance phải tách riêng, tenant A không được thấy process của tenant B. Activiti hỗ trợ sẵn multi-tenancy qua tenantId.

Có điều tenantId chỉ là một cột để lọc, engine không tự chặn gì cả. Query nào quên ...TenantId(...) là thấy hết data của mọi tenant. Ngược lại, definition đã deploy kèm tenant thì startProcessInstanceByKey() (bản không có tenant) sẽ báo không tìm thấy, phải dùng startProcessInstanceByKeyAndTenantId(). Nên gom mọi chỗ gọi engine vào một lớp service có sẵn tenant như dưới đây, đừng để mỗi nơi tự nhớ.

@Service
@RequiredArgsConstructor
public class TenantAwareProcessService {
 
   private final RepositoryService repositoryService;
   private final RuntimeService runtimeService;
   private final TaskService taskService;
 
   // Deploy với tenant isolation
   public String deploy(String bpmnXml, String name, String workspaceId) {
       Deployment deployment = repositoryService.createDeployment()
           .name(name)
           .tenantId(workspaceId)  // ← Tenant isolation
           .addString(name + ".bpmn20.xml", bpmnXml)
           .deploy();
       return deployment.getId();
   }
 
   // Start instance — tự động inherit tenant từ definition
   public String startProcess(String processKey, String workspaceId, Map<String, Object> vars) {
       ProcessInstance instance = runtimeService.startProcessInstanceByKeyAndTenantId(
           processKey, vars, workspaceId);  // ← Tenant-scoped
       return instance.getId();
   }
 
   // Query tasks — chỉ trong workspace
   public List<Task> getTasks(String userId, String workspaceId) {
       return taskService.createTaskQuery()
           .taskAssignee(userId)
           .taskTenantId(workspaceId)  // ← Tenant filter
           .list();
   }
}

# testing processes

Test BPMN process là test cả luồng (diagram) lẫn business logic (delegate). Delegate thì unit test riêng, còn cả process thì integration test.

Hai điều cần để ý với test bên dưới. Thứ nhất, test kiểu "start xong là kiểm tra process đã kết thúc" chỉ đúng khi các service task trên đường đi đều sync. Task nào async="true" thì lúc startProcessInstanceByKey() trả về, job mới chỉ nằm trong DB; phải chờ job executor chạy xong (poll có timeout) hoặc tắt async-executor-activate trong profile test rồi tự gọi managementService.executeJob(jobId). Thứ hai, @MockBean đã deprecated từ Spring Boot 3.4 và bị bỏ ở Boot 4; dự án dùng Boot 3.4 trở lên thì đổi sang @MockitoBean.

@SpringBootTest
@ActiveProfiles("test")
class LoanApprovalProcessTest {
 
   @Autowired private RuntimeService runtimeService;
   @Autowired private TaskService taskService;
   @Autowired private HistoryService historyService;
   @Autowired private RepositoryService repositoryService;
 
   @MockBean private CreditScoringClient creditScoringClient;
   @MockBean private NotificationService notificationService;
 
   @BeforeEach
   void deployProcess() {
       repositoryService.createDeployment()
           .addClasspathResource("processes/loan-approval.bpmn20.xml")
           .deploy();
   }
 
   @Test
   void loanApproval_highCreditScore_autoApproved() {
       // Given: high credit score → auto-approve path
       when(creditScoringClient.calculate(any(), any()))
           .thenReturn(new CreditScoreResult(800, RiskLevel.LOW));
 
       // When: start process
       Map<String, Object> variables = Map.of(
           "customerId", "CUST-001",
           "loanAmount", new BigDecimal("500000000"),
           "applicantName", "Nguyen Van A"
       );
       ProcessInstance instance = runtimeService.startProcessInstanceByKey(
           "loan-approval", "LOAN-001", variables);
 
       // Then: process should complete automatically (no user task needed)
       HistoricProcessInstance completed = historyService.createHistoricProcessInstanceQuery()
           .processInstanceId(instance.getId())
           .finished()
           .singleResult();
 
       assertThat(completed).isNotNull();
       assertThat(completed.getEndTime()).isNotNull();
 
       // Verify notification sent
       verify(notificationService).sendApprovalNotification("CUST-001", "LOAN-001");
   }
 
   @Test
   void loanApproval_lowCreditScore_requiresManualReview() {
       // Given: low credit score → manual review path
       when(creditScoringClient.calculate(any(), any()))
           .thenReturn(new CreditScoreResult(550, RiskLevel.HIGH));
 
       // When: start process
       ProcessInstance instance = runtimeService.startProcessInstanceByKey(
           "loan-approval", "LOAN-002", Map.of(
               "customerId", "CUST-002",
               "loanAmount", new BigDecimal("2000000000")
           ));
 
       // Then: should create user task for manual review
       Task reviewTask = taskService.createTaskQuery()
           .processInstanceId(instance.getId())
           .taskDefinitionKey("manualReviewTask")
           .singleResult();
 
       assertThat(reviewTask).isNotNull();
       assertThat(reviewTask.getName()).isEqualTo("Manual Review");
 
       // Complete review with approval
       taskService.complete(reviewTask.getId(), Map.of(
           "decision", "approve",
           "reviewComments", "High amount but good collateral"
       ));
 
       // Verify process completed
       HistoricProcessInstance completed = historyService.createHistoricProcessInstanceQuery()
           .processInstanceId(instance.getId())
           .finished()
           .singleResult();
       assertThat(completed).isNotNull();
   }
 
   @Test
   void loanApproval_scoringServiceDown_errorPath() {
       // Given: scoring service throws error
       when(creditScoringClient.calculate(any(), any()))
           .thenThrow(new ScoringServiceUnavailableException("Service down"));
 
       // When: start process
       ProcessInstance instance = runtimeService.startProcessInstanceByKey(
           "loan-approval", "LOAN-003", Map.of(
               "customerId", "CUST-003",
               "loanAmount", new BigDecimal("100000000")
           ));
 
       // Then: should route to error handling path (manual scoring task)
       Task manualScoringTask = taskService.createTaskQuery()
           .processInstanceId(instance.getId())
           .taskDefinitionKey("manualScoringTask")
           .singleResult();
 
       assertThat(manualScoringTask).isNotNull();
       assertThat(manualScoringTask.getName()).isEqualTo("Manual Credit Assessment");
   }
}

# monitoring và xử lý sự cố trên production

# nên theo dõi metric nào

@Component
@RequiredArgsConstructor
public class ProcessMetricsCollector {
 
   private final RuntimeService runtimeService;
   private final ManagementService managementService;
   private final TaskService taskService;
   private final MeterRegistry meterRegistry;
 
   // Gauge giữ tham chiếu tới object, nên phải là AtomicLong sống suốt vòng đời bean
   private final AtomicLong activeInstances = new AtomicLong();
   private final AtomicLong suspendedInstances = new AtomicLong();
   private final AtomicLong pendingTasks = new AtomicLong();
   private final AtomicLong asyncJobs = new AtomicLong();
   private final AtomicLong deadLetterJobs = new AtomicLong();
   private final AtomicLong timerJobs = new AtomicLong();
 
   @PostConstruct
   void registerGauges() {
       meterRegistry.gauge("bpmn.instances.active", activeInstances);
       meterRegistry.gauge("bpmn.instances.suspended", suspendedInstances);
       meterRegistry.gauge("bpmn.tasks.pending", pendingTasks);
       meterRegistry.gauge("bpmn.jobs.async", asyncJobs);
       meterRegistry.gauge("bpmn.jobs.deadletter", deadLetterJobs);
       meterRegistry.gauge("bpmn.jobs.timer", timerJobs);
   }
 
   @Scheduled(fixedRate = 30000)  // Every 30s, cần @EnableScheduling
   public void collectMetrics() {
       // Active instances
       activeInstances.set(runtimeService.createProcessInstanceQuery().active().count());
 
       // Suspended instances (paused, needs attention)
       suspendedInstances.set(runtimeService.createProcessInstanceQuery().suspended().count());
 
       // Pending user tasks (work backlog)
       pendingTasks.set(taskService.createTaskQuery().count());
 
       // Async jobs đang chờ executor (tăng đều là executor không theo kịp)
       asyncJobs.set(managementService.createJobQuery().count());
 
       // Failed jobs (needs intervention)
       deadLetterJobs.set(managementService.createDeadLetterJobQuery().count());
 
       // Timer jobs waiting (scheduled future work + job fail đang chờ retry)
       timerJobs.set(managementService.createTimerJobQuery().count());
   }
}

# lỗi hay gặp và cách xử lý

SymptomRoot CauseSolution
Process stuck at service taskDelegate throws uncaught exceptionXem EXCEPTION_MSG_ trong ACT_RU_TIMER_JOB / ACT_RU_DEADLETTER_JOB, fix delegate, retry
User task not appearingWrong assignee/candidateGroupCheck BPMN definition, verify user in correct group
Process very slowSynchronous external callsMake service tasks async
"Optimistic locking exception"Concurrent updates on same instanceUse exclusive="true" on async tasks
History tables hugeNo cleanup/archivingSchedule history cleanup job
Deadletter jobs piling upExternal service permanently downFix service, then moveDeadLetterJobToExecutable

# kết luận

Activiti mạnh nhưng phải học một thời gian mới quen. Mấy điều mình rút ra:

  • BPMN diagram cũng là code chạy thật: đối xử với nó như code production (version, test, review)
  • Delegate là code của bạn: giữ cho nhỏ, idempotent, dễ test
  • I/O thì để async: đừng để external call block thread của engine
  • Canh deadletter job: lỗi âm thầm kiểu này mới là thứ làm sập production
  • Làm multi-tenancy ngay từ đầu: để sau mới chắp tenant isolation vào là khổ
  • Test cả luồng: unit test delegate + integration test các nhánh của process
  • History là mỏ vàng: dùng cho analytics, tìm chỗ nghẽn, compliance

Chỉ là những ghi chép cá nhân với hy vọng mang lại chút giá trị. Nếu thấy hữu ích, đừng ngại chia sẻ cho bạn bè & đồng nghiệp nhé!

Happy coding 😎 👍🏻 🚀 🔥.