activiti 8
Tung Bui Thanh (TungDaDev)22 mins read

# 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 8 | Camunda 7/8 | Flowable 7 | |
|---|---|---|---|
| License | Apache 2.0 | Community + Enterprise | Apache 2.0 |
| Cloud-native | Activiti Cloud (K8s) | Camunda 8 (Zeebe) | Flowable Cloud |
| Spring Boot | Native integration | Good integration | Native integration |
| Runtime model | Shared DB (RDBMS) | Shared DB / Event-sourced (C8) | Shared DB |
| Ghi chú | Bản cloud của Alfresco | C7 CE hết hỗ trợ từ 10/2025; C8 tự host cần license cho production | Fork 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:
| Prefix | Tên | Mục đích | Ví dụ |
|---|---|---|---|
ACT_RE_ | Repository | Process definitions, deployments | ACT_RE_PROCDEF, ACT_RE_DEPLOYMENT |
ACT_RU_ | Runtime | Running instances, tasks, variables, jobs | ACT_RU_EXECUTION, ACT_RU_TASK |
ACT_HI_ | History | Completed instances, audit trail | ACT_HI_PROCINST, ACT_HI_ACTINST |
ACT_GE_ | General | Byte arrays, properties | ACT_GE_BYTEARRAY |
ACT_ID_ | Identity | Users, 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:
- Single Responsibility: 1 delegate làm đúng một việc. Đừng nhồi 500 dòng logic vào 1 delegate.
- Idempotent: engine có thể retry, nên
chargePayment()phải check "charge chưa?" rồi mới charge. - 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.
- 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?
| Scenario | Sync | Async | Lý 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ý
| Symptom | Root Cause | Solution |
|---|---|---|
| Process stuck at service task | Delegate throws uncaught exception | Xem EXCEPTION_MSG_ trong ACT_RU_TIMER_JOB / ACT_RU_DEADLETTER_JOB, fix delegate, retry |
| User task not appearing | Wrong assignee/candidateGroup | Check BPMN definition, verify user in correct group |
| Process very slow | Synchronous external calls | Make service tasks async |
| "Optimistic locking exception" | Concurrent updates on same instance | Use exclusive="true" on async tasks |
| History tables huge | No cleanup/archiving | Schedule history cleanup job |
| Deadletter jobs piling up | External service permanently down | Fix 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 😎 👍🏻 🚀 🔥.