SpringBoot整合Activiti工作流的完整教程

springboot-activiti-demo
├── src/main/java/com/example/activiti
│   ├── config/                    # 配置类
│   │   ├── ActivitiConfig.java    # Activiti配置（性能优化）
│   │   ├── MailConfig.java        # 邮件配置
│   │   ├── AsyncConfig.java       # 异步配置
│   │   └── CacheConfig.java       # 缓存配置
│   ├── controller/                # 控制器
│   │   ├── LeaveController.java
│   │   └── WorkflowController.java
│   ├── service/                   # 服务层
│   │   ├── WorkflowService.java
│   │   ├── NotificationService.java
│   │   ├── LeaveService.java
│   │   ├── CacheService.java      # 缓存服务
│   │   └── impl/
│   │       ├── WorkflowServiceImpl.java
│   │       ├── NotificationServiceImpl.java
│   │       ├── LeaveServiceImpl.java
│   │       └── CacheServiceImpl.java
│   ├── entity/                    # 实体类
│   │   ├── LeaveRequest.java
│   │   ├── TaskInfo.java
│   │   ├── HistoryInfo.java
│   │   └── NotificationRecord.java
│   ├── mapper/                    # 数据访问层
│   │   ├── LeaveRequestMapper.java
│   │   └── NotificationRecordMapper.java
│   ├── delegate/                  # 委托类（服务任务）
│   │   └── HRArchiveTask.java
│   ├── listener/                  # 监听器（异步优化）
│   │   ├── ProcessStartListener.java
│   │   ├── TaskCreateListener.java
│   │   └── TaskCompleteListener.java
│   ├── async/                     # 异步处理器
│   │   └── NotificationAsyncProcessor.java
│   └── ActivitiApplication.java  # 启动类
├── src/main/resources
│   ├── processes/                # BPMN流程定义文件目录
│   │   └── leave-request.bpmn20.xml
│   ├── mapper/                   # MyBatis Mapper XML
│   │   ├── LeaveRequestMapper.xml
│   │   └── NotificationRecordMapper.xml
│   ├── templates/                # 模板文件（带缓存）
│   │   └── email/                 # 邮件模板
│   │       ├── process-started.html
│   │       ├── task-created.html
│   │       └── task-completed.html
│   └── application.yml          # 配置文件（性能优化）
└── pom.xml                       # Maven依赖

-- 创建数据库，使用InnoDB引擎以获得更好的事务性能
CREATE DATABASE IF NOT EXISTS activiti_demo 
CHARACTER SET utf8mb4 
COLLATE utf8mb4_unicode_ci;
-- 使用数据库
USE activiti_demo;

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.18</version>
        <relativePath/>
    </parent>
    <groupId>com.example</groupId>
    <artifactId>springboot-activiti-demo</artifactId>
    <version>1.0.0</version>
    <name>SpringBoot Activiti Demo</name>
    <description>SpringBoot整合Activiti工作流引擎完整示例</description>
    <properties>
        <java.version>1.8</java.version>
        <activiti.version>7.1.0.M6</activiti.version>
        <mysql.version>8.0.33</mysql.version>
        <druid.version>1.2.18</druid.version>
        <mybatis-plus.version>3.5.5</mybatis-plus.version>
        <caffeine.version>3.1.8</caffeine.version>
    </properties>
    <dependencies>
        <!-- SpringBoot Web Starter -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- SpringBoot Validation -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-validation</artifactId>
        </dependency>
        <!-- SpringBoot Mail（邮件通知功能） -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-mail</artifactId>
        </dependency>
        <!-- SpringBoot Thymeleaf（邮件模板引擎） -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-thymeleaf</artifactId>
        </dependency>
        <!-- Activiti SpringBoot Starter -->
        <dependency>
            <groupId>org.activiti</groupId>
            <artifactId>activiti-spring-boot-starter</artifactId>
            <version>${activiti.version}</version>
        </dependency>
        <!-- MySQL 驱动 -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>${mysql.version}</version>
        </dependency>
        <!-- Druid 数据库连接池（高性能） -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>${druid.version}</version>
        </dependency>
        <!-- MyBatis-Plus -->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>${mybatis-plus.version}</version>
        </dependency>
        <!-- Caffeine 高性能缓存 -->
        <dependency>
            <groupId>com.github.ben-manes.caffeine</groupId>
            <artifactId>caffeine</artifactId>
            <version>${caffeine.version}</version>
        </dependency>
        <!-- Spring Cache -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-cache</artifactId>
        </dependency>
        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <!-- Hutool工具类 -->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.26</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

server:
  port: 8080
  tomcat:
    # Tomcat线程池优化
    threads:
      max: 200          # 最大工作线程数
      min-spare: 20     # 最小空闲线程数
    max-connections: 10000  # 最大连接数
    accept-count: 100   # 等待队列长度
spring:
  application:
    name: springboot-activiti-demo
  datasource:
    url: jdbc:mysql://localhost:3306/activiti_demo?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai&nullCatalogMeansCurrent=true&useSSL=false
    username: root
    password: root
    driver-class-name: com.mysql.cj.jdbc.Driver
    type: com.alibaba.druid.pool.DruidDataSource
    druid:
      # 连接池基础配置
      initial-size: 10                # 初始连接数
      min-idle: 10                   # 最小空闲连接数
      max-active: 100                 # 最大活跃连接数
      max-wait: 60000                # 获取连接最大等待时间（毫秒）
      # 连接检测配置
      test-while-idle: true          # 检测空闲连接是否有效
      test-on-borrow: false          # 获取连接时不检测（避免性能损耗）
      test-on-return: false          # 归还连接时不检测
      validation-query: SELECT 1     # 验证SQL
      time-between-eviction-runs-millis: 60000    # 检测空闲连接间隔
      min-evictable-idle-time-millis: 300000     # 空闲连接最小生存时间
      # 连接泄漏保护
      remove-abandoned: true         # 是否移除泄漏连接
      remove-abandoned-timeout: 180000 # 泄漏连接超时时间（秒）
      log-abandoned: true            # 记录泄漏连接日志
      # 性能监控（生产环境建议关闭）
      stat-view-servlet:
        enabled: false
      filter:
        stat:
          enabled: false
        wall:
          enabled: true
          config:
            multi-statement-allow: true  # 允许批量SQL
  # 缓存配置
  cache:
    type: caffeine
    caffeine:
      spec: maximumSize=1000,expireAfterWrite=30m
  # 邮件配置
  mail:
    host: smtp.qq.com
    port: 587
    username: your-email@qq.com
    password: your-authorization-code
    protocol: smtp
    default-encoding: UTF-8
    properties:
      mail:
        smtp:
          auth: true
          starttls:
            enable: true
          ssl:
            trust: smtp.qq.com
          # 连接池配置
          connectiontimeout: 10000
          timeout: 10000
          writetimeout: 10000
    from: your-email@qq.com
    # 异步发送配置
    async: true
  # Thymeleaf模板引擎配置
  thymeleaf:
    prefix: classpath:/templates/
    suffix: .html
    mode: HTML
    encoding: UTF-8
    cache: true  # 生产环境开启缓存，提升性能
    servlet:
      content-type: text/html
  # 异步任务配置
  task:
    execution:
      pool:
        core-size: 10               # 核心线程数
        max-size: 50               # 最大线程数
        queue-capacity: 1000       # 队列容量
        keep-alive: 60s            # 线程空闲时间
      thread-name-prefix: async-   # 线程名前缀
    scheduling:
      pool:
        size: 5                    # 定时任务线程池大小
    shutdown:
      await-termination: true
      await-termination-period: 60s
# MyBatis-Plus配置
mybatis-plus:
  mapper-locations: classpath:mapper/*.xml
  type-aliases-package: com.example.activiti.entity
  configuration:
    map-underscore-to-camel-case: true
    # 生产环境关闭SQL日志，提升性能
    # log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
    # 二级缓存配置
    cache-enabled: true
    lazy-loading-enabled: true    # 开启懒加载
    default-executor-type: REUSE  # 复用预编译语句
  global-config:
    db-config:
      # 性能优化：批量操作
      batch-insert: true
      batch-update: true
# Activiti 配置（性能优化版）
activiti:
  # 数据库配置
  database-schema-update: true
  database-type: mysql
  # 历史记录配置：使用ACTIVITY而非FULL，减少数据存储量，提升性能
  history-level: activity
  # 流程定义部署模式
  deployment-mode: single-engine
  # 流程定义缓存配置（性能优化关键）
  process-definition-cache-limit: 500      # 增加缓存数量
  enable-process-definition-info-cache: true  # 启用流程定义信息缓存
  process-definition-cache-enabled: true
  # 异步执行器配置（性能提升关键）
  async-executor-activate: true
  async-executor-enabled: true
  async-executor-default-async-job-acquire-wait-time: 10000    # 默认异步任务获取等待时间
  async-executor-default-async-job-acquire-lock-time: 300000    # 默认异步任务获取锁时间
  async-executor-default-queue-size-full-wait-time: 50         # 队列满时等待时间
  async-executor-maximum-async-jobs-due-per-acquisition: 100  # 单次获取最大任务数
  # 流程实例缓存
  enable-process-instance-info-cache: true
  # 批量操作配置
  process-instance-query-limit: 100      # 流程实例查询限制
  task-query-limit: 100                  # 任务查询限制
  batch-operations-size: 50              # 批量操作大小
  # 安全配置
  security:
    enabled: false
  # 流程定义部署路径
  process-definition-location-prefix: classpath*:/processes/
  process-definition-location-suffixes: 
    - **.bpmn20.xml
    - **.bpmn
  # 作业执行器配置
  job-executor-activate: true
  async-history-enabled: true            # 启用异步历史记录（性能优化）
# 消息通知配置
notification:
  email-enabled: true
  system-enabled: true
  # 异步发送配置
  async-enabled: true
  # 批量发送配置
  batch-size: 50
  # 重试配置
  retry-count: 3
  retry-interval: 5000                  # 重试间隔（毫秒）
  types: PROCESS_START,TASK_CREATE,TASK_COMPLETE,PROCESS_END
# Caffeine缓存配置
caffeine:
  cache-specs:
    # 流程定义缓存
    process-definition: maximumSize=200,expireAfterWrite=1h
    # 用户信息缓存
    user-info: maximumSize=1000,expireAfterWrite=30m
    # 邮件模板缓存
    email-template: maximumSize=100,expireAfterWrite=24h
    # 流程变量缓存
    process-variable: maximumSize=5000,expireAfterWrite=10m
# 日志配置（性能优化：异步日志）
logging:
  pattern:
    console: '%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{50} - %msg%n'
  level:
    root: INFO
    org.activiti: WARN                 # 降低Activiti日志级别
    com.alibaba.druid: WARN
    com.example.activiti: INFO
    # 异步日志配置
  async:
    enabled: true
    queue-capacity: 10000
    discard-threshold: 0
    include-caller-data: false        # 不包含调用者数据，提升性能

-- 请假申请表
CREATE TABLE t_leave_request (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主键ID',
    applicant_id VARCHAR(64) NOT NULL COMMENT '申请人ID',
    applicant_name VARCHAR(100) NOT NULL COMMENT '申请人姓名',
    applicant_email VARCHAR(255) COMMENT '申请人邮箱',
    leave_type INT NOT NULL COMMENT '请假类型：1-年假，2-事假，3-病假，4-婚假，5-产假，6-丧假',
    start_time DATETIME NOT NULL COMMENT '请假开始时间',
    end_time DATETIME NOT NULL COMMENT '请假结束时间',
    leave_days DOUBLE NOT NULL COMMENT '请假天数',
    reason TEXT COMMENT '请假事由',
    process_instance_id VARCHAR(64) COMMENT '流程实例ID',
    status INT DEFAULT 0 COMMENT '审批状态：0-草稿，1-待审批，2-已批准，3-已拒绝，4-已取消，5-已归档',
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    deleted INT DEFAULT 0 COMMENT '逻辑删除标记',
    -- 索引优化：根据实际查询场景建立索引
    INDEX idx_applicant_id (applicant_id) COMMENT '申请人索引',
    INDEX idx_process_instance_id (process_instance_id) COMMENT '流程实例ID索引',
    INDEX idx_status (status) COMMENT '状态索引',
    INDEX idx_create_time (create_time) COMMENT '创建时间索引',
    INDEX idx_applicant_status (applicant_id, status) COMMENT '联合索引',
    -- 引擎优化：使用InnoDB引擎，支持事务和外键
    ENGINE=InnoDB 
    DEFAULT CHARSET=utf8mb4 
    COLLATE=utf8mb4_unicode_ci 
    COMMENT='请假申请表'
) ROW_FORMAT=DYNAMIC;  -- 使用DYNAMIC行格式，支持大字段

-- 消息通知记录表
CREATE TABLE t_notification_record (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主键ID',
    notification_type VARCHAR(50) NOT NULL COMMENT '通知类型',
    process_instance_id VARCHAR(64) COMMENT '流程实例ID',
    process_definition_id VARCHAR(64) COMMENT '流程定义ID',
    task_id VARCHAR(64) COMMENT '任务ID',
    task_name VARCHAR(255) COMMENT '任务名称',
    recipient_id VARCHAR(64) COMMENT '接收人ID',
    recipient_name VARCHAR(255) COMMENT '接收人姓名',
    recipient_email VARCHAR(255) COMMENT '接收人邮箱',
    sender_id VARCHAR(64) COMMENT '发送人ID',
    sender_name VARCHAR(255) COMMENT '发送人姓名',
    notification_title VARCHAR(500) COMMENT '通知标题',
    notification_content TEXT COMMENT '通知内容',
    notification_status VARCHAR(20) DEFAULT 'PENDING' COMMENT '通知状态',
    send_time DATETIME COMMENT '发送时间',
    error_message TEXT COMMENT '错误信息',
    retry_count INT DEFAULT 0 COMMENT '重试次数',
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    -- 索引优化
    INDEX idx_process_instance_id (process_instance_id) COMMENT '流程实例ID索引',
    INDEX idx_recipient_id (recipient_id) COMMENT '接收人ID索引',
    INDEX idx_notification_status (notification_status) COMMENT '状态索引',
    INDEX idx_create_time (create_time) COMMENT '创建时间索引',
    INDEX idx_type_status (notification_type, notification_status) COMMENT '联合索引',
    ENGINE=InnoDB 
    DEFAULT CHARSET=utf8mb4 
    COLLATE=utf8mb4_unicode_ci 
    COMMENT='消息通知记录表'
) ROW_FORMAT=DYNAMIC;

-- 为Activiti运行时表添加索引（提升查询性能）
-- 任务表索引
CREATE INDEX idx_act_ru_task_assignee ON ACT_RU_TASK(ASSIGNEE_);
CREATE INDEX idx_act_ru_task_proc_inst ON ACT_RU_TASK(PROC_INST_ID_);
CREATE INDEX idx_act_ru_task_create_time ON ACT_RU_TASK(CREATE_TIME_);
-- 变量表索引
CREATE INDEX idx_act_ru_var_proc_inst ON ACT_RU_VARIABLE(PROC_INST_ID_);
CREATE INDEX idx_act_ru_var_name ON ACT_RU_VARIABLE(NAME_);
-- 历史任务表索引
CREATE INDEX idx_act_hi_task_inst_proc_inst ON ACT_HI_TASKINST(PROC_INST_ID_);
CREATE INDEX idx_act_hi_task_inst_assignee ON ACT_HI_TASKINST(ASSIGNEE_);
CREATE INDEX idx_act_hi_task_inst_end_time ON ACT_HI_TASKINST(END_TIME_);
-- 历史变量表索引
CREATE INDEX idx_act_hi_var_inst_proc_inst ON ACT_HI_VARINST(PROC_INST_ID_);
CREATE INDEX idx_act_hi_var_inst_name ON ACT_HI_VARINST(NAME_);
-- 优化表存储引擎（确保使用InnoDB）
ALTER TABLE ACT_RU_TASK ENGINE=InnoDB;
ALTER TABLE ACT_RU_VARIABLE ENGINE=InnoDB;
ALTER TABLE ACT_HI_TASKINST ENGINE=InnoDB;
ALTER TABLE ACT_HI_VARINST ENGINE=InnoDB;

package com.example.activiti.entity;
import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
import java.io.Serializable;
import java.util.Date;
/**
 * 任务信息DTO
 * 使用Serializable接口，支持缓存和序列化
 */
@Data
public class TaskInfo implements Serializable {
    private static final long serialVersionUID = 1L;
    private String taskId;
    private String taskName;
    private String taskDescription;
    private String assignee;
    private String candidateGroups;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private Date createTime;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private Date dueDate;
    private Integer priority;
    private String processInstanceId;
    private String processDefinitionId;
    private String processDefinitionName;
    private String processDefinitionKey;
    private String businessKey;
    private String taskKey;
}

package com.example.activiti.entity;

import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
import java.io.Serializable;
import java.util.Date;

/**
 * 历史信息DTO
 */
@Data
public class HistoryInfo implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private String taskId;
    private String taskName;
    private String taskDescription;
    private String assignee;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private Date startTime;
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private Date endTime;
    private Long duration;
    private String processInstanceId;
    private String processDefinitionId;
    private String deleteReason;
    private String taskKey;
}

package com.example.activiti.entity;

import com.baomidou.mybatisplus.annotation.*;
import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
import java.io.Serializable;
import java.util.Date;

/**
 * 消息通知记录实体
 */
@Data
@TableName("t_notification_record")
public class NotificationRecord implements Serializable {
    
    private static final long serialVersionUID = 1L;
    
    public enum NotificationType {
        PROCESS_START, TASK_CREATE, TASK_COMPLETE, PROCESS_END
    }
    
    public enum NotificationStatus {
        PENDING, SENT, FAILED
    }
    
    @TableId(value = "id", type = IdType.AUTO)
    private Long id;
    
    private String notificationType;
    private String processInstanceId;
    private String processDefinitionId;
    private String taskId;
    private String taskName;
    private String recipientId;
    private String recipientName;
    private String recipientEmail;
    private String senderId;
    private String senderName;
    private String notificationTitle;
    private String notificationContent;
    private String notificationStatus;
    
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private Date sendTime;
    
    private String errorMessage;
    private Integer retryCount;
    
    @TableField(fill = FieldFill.INSERT)
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private Date createTime;
    
    @TableField(fill = FieldFill.INSERT_UPDATE)
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private Date updateTime;
}

package com.example.activiti.service;

import com.example.activiti.entity.HistoryInfo;
import com.example.activiti.entity.TaskInfo;
import org.activiti.engine.repository.Deployment;
import org.activiti.engine.runtime.ProcessInstance;

import java.util.List;
import java.util.Map;

public interface WorkflowService {
    
    // ========== 流程定义管理 ==========
    
    /**
     * 部署流程定义（支持批量部署）
     * @param resourceName 资源名称
     * @param bpmnContent BPMN内容
     * @return 部署信息
     */
    Deployment deployProcess(String resourceName, String bpmnContent);
    
    /**
     * 批量部署流程定义（性能优化）
     * @param resources 资源列表
     * @return 部署信息列表
     */
    List<Deployment> deployProcesses(Map<String, String> resources);
    
    /**
     * 删除流程定义（级联删除）
     * @param deploymentId 部署ID
     * @param cascade 是否级联删除
     */
    void deleteDeployment(String deploymentId, boolean cascade);
    
    // ========== 流程实例管理 ==========
    
    /**
     * 启动流程实例（支持流程变量批量设置）
     * @param processDefinitionKey 流程定义Key
     * @param businessKey 业务Key
     * @param variables 流程变量
     * @return 流程实例
     */
    ProcessInstance startProcess(String processDefinitionKey, String businessKey, 
                                   Map<String, Object> variables);
    
    /**
     * 批量启动流程实例（性能优化）
     * @param processDefinitionKey 流程定义Key
     * @param businessKeys 业务Key列表
     * @param variablesList 流程变量列表
     * @return 流程实例列表
     */
    List<ProcessInstance> startProcesses(String processDefinitionKey, 
                                          List<String> businessKeys, 
                                          List<Map<String, Object>> variablesList);
    
    /**
     * 删除流程实例
     * @param processInstanceId 流程实例ID
     * @param deleteReason 删除原因
     */
    void deleteProcessInstance(String processInstanceId, String deleteReason);
    
    /**
     * 批量删除流程实例（性能优化）
     * @param processInstanceIds 流程实例ID列表
     * @param deleteReason 删除原因
     */
    void batchDeleteProcessInstances(List<String> processInstanceIds, String deleteReason);
    
    /**
     * 挂起流程实例
     * @param processInstanceId 流程实例ID
     */
    void suspendProcessInstance(String processInstanceId);
    
    /**
     * 激活流程实例
     * @param processInstanceId 流程实例ID
     */
    void activateProcessInstance(String processInstanceId);
    
    // ========== 任务管理 ==========
    
    /**
     * 查询用户待办任务（支持分页和排序）
     * @param userId 用户ID
     * @param page 页码
     * @param size 每页大小
     * @return 任务列表
     */
    List<TaskInfo> getUserTasks(String userId, int page, int size);
    
    /**
     * 查询用户待办任务（高性能版本）
     * @param userId 用户ID
     * @param candidateGroups 候选组列表
     * @return 任务列表
     */
    List<TaskInfo> getUserTasksOptimized(String userId, List<String> candidateGroups);
    
    /**
     * 查询用户已办任务（支持分页）
     * @param userId 用户ID
     * @param page 页码
     * @param size 每页大小
     * @return 历史任务列表
     */
    List<HistoryInfo> getUserCompletedTasks(String userId, int page, int size);
    
    /**
     * 完成任务（支持批量变量设置）
     * @param taskId 任务ID
     * @param variables 流程变量
     */
    void completeTask(String taskId, Map<String, Object> variables);
    
    /**
     * 批量完成任务（性能优化）
     * @param taskIds 任务ID列表
     * @param variables 流程变量
     */
    void batchCompleteTasks(List<String> taskIds, Map<String, Object> variables);
    
    /**
     * 认领任务
     * @param taskId 任务ID
     * @param userId 用户ID
     */
    void claimTask(String taskId, String userId);
    
    /**
     * 转办任务
     * @param taskId 任务ID
     * @param targetUserId 目标用户ID
     */
    void transferTask(String taskId, String targetUserId);
    
    /**
     * 委派任务
     * @param taskId 任务ID
     * @param targetUserId 目标用户ID
     */
    void delegateTask(String taskId, String targetUserId);
    
    // ========== 流程变量管理 ==========
    
    /**
     * 获取流程变量（支持缓存）
     * @param processInstanceId 流程实例ID
     * @param variableName 变量名
     * @return 变量值
     */
    Object getVariable(String processInstanceId, String variableName);
    
    /**
     * 批量获取流程变量（性能优化）
     * @param processInstanceId 流程实例ID
     * @param variableNames 变量名列表
     * @return 变量Map
     */
    Map<String, Object> getVariables(String processInstanceId, List<String> variableNames);
    
    /**
     * 设置流程变量（批量设置）
     * @param processInstanceId 流程实例ID
     * @param variables 变量Map
     */
    void setVariables(String processInstanceId, Map<String, Object> variables);
    
    /**
     * 批量设置流程变量（性能优化）
     * @param processInstanceIds 流程实例ID列表
     * @param variableName 变量名
     * @param variableValue 变量值
     */
    void batchSetVariables(List<String> processInstanceIds, String variableName, Object variableValue);
    
    // ========== 流程历史管理 ==========
    
    /**
     * 查询流程历史（支持分页）
     * @param processInstanceId 流程实例ID
     * @param page 页码
     * @param size 每页大小
     * @return 历史任务列表
     */
    List<HistoryInfo> getProcessHistory(String processInstanceId, int page, int size);
    
    /**
     * 查询流程变量历史
     * @param processInstanceId 流程实例ID
     * @return 变量历史列表
     */
    List<Map<String, Object>> getVariableHistory(String processInstanceId);
}

package com.example.activiti.service.impl;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.example.activiti.entity.HistoryInfo;
import com.example.activiti.entity.TaskInfo;
import com.example.activiti.service.CacheService;
import com.example.activiti.service.WorkflowService;
import lombok.extern.slf4j.Slf4j;
import org.activiti.engine.*;
import org.activiti.engine.history.HistoricTaskInstance;
import org.activiti.engine.repository.Deployment;
import org.activiti.engine.runtime.ProcessInstance;
import org.activiti.engine.task.Task;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.ByteArrayInputStream;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 工作流服务实现类（性能优化版）
 */
@Slf4j
@Service
public class WorkflowServiceImpl implements WorkflowService {
    
    @Autowired
    private RepositoryService repositoryService;
    
    @Autowired
    private RuntimeService runtimeService;
    
    @Autowired
    private TaskService taskService;
    
    @Autowired
    private HistoryService historyService;
    
    @Autowired
    private ManagementService managementService;
    
    @Autowired
    private CacheService cacheService;
    
    // ========== 流程定义管理 ==========
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public Deployment deployProcess(String resourceName, String bpmnContent) {
        log.debug("部署流程定义：resourceName={}", resourceName);
        
        // 清除流程定义缓存
        clearProcessDefinitionCache(resourceName);
        
        Deployment deployment = repositoryService.createDeployment()
                .addInputStream(resourceName, new ByteArrayInputStream(bpmnContent.getBytes()))
                .deploy();
        
        log.debug("流程定义部署成功：deploymentId={}", deployment.getId());
        return deployment;
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public List<Deployment> deployProcesses(Map<String, String> resources) {
        log.debug("批量部署流程定义，数量：{}", resources.size());
        
        DeploymentBuilder builder = repositoryService.createDeployment();
        resources.forEach((name, content) -> {
            builder.addInputStream(name, new ByteArrayInputStream(content.getBytes()));
            // 清除缓存
            clearProcessDefinitionCache(name);
        });
        
        Deployment deployment = builder.deploy();
        
        // 返回所有部署信息
        return repositoryService.createDeploymentQuery()
                .deploymentId(deployment.getId())
                .list();
    }
    
    /**
     * 清除流程定义缓存
     */
    private void clearProcessDefinitionCache(String resourceName) {
        try {
            String processKey = resourceName.replace(".bpmn20.xml", "").replace(".bpmn", "");
            cacheService.evictProcessDefinitionCache(processKey);
        } catch (Exception e) {
            log.warn("清除流程定义缓存失败", e);
        }
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void deleteDeployment(String deploymentId, boolean cascade) {
        log.debug("删除流程定义：deploymentId={}, cascade={}", deploymentId, cascade);
        
        // 获取流程定义信息，清除缓存
        if (cascade) {
            List<org.activiti.engine.repository.ProcessDefinition> definitions = 
                    repositoryService.createProcessDefinitionQuery()
                            .deploymentId(deploymentId)
                            .list();
            definitions.forEach(def -> {
                cacheService.evictProcessDefinitionCache(def.getKey());
            });
        }
        
        repositoryService.deleteDeployment(deploymentId, cascade);
    }
    
    // ========== 流程实例管理 ==========
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public ProcessInstance startProcess(String processDefinitionKey, String businessKey, 
                                          Map<String, Object> variables) {
        log.debug("启动流程实例：processDefinitionKey={}, businessKey={}", 
                  processDefinitionKey, businessKey);
        
        ProcessInstance processInstance = runtimeService.startProcessInstanceByKey(
                processDefinitionKey, businessKey, variables);
        
        log.debug("流程实例启动成功：processInstanceId={}", processInstance.getId());
        return processInstance;
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public List<ProcessInstance> startProcesses(String processDefinitionKey, 
                                                  List<String> businessKeys, 
                                                  List<Map<String, Object>> variablesList) {
        log.debug("批量启动流程实例，数量：{}", businessKeys.size());
        
        List<ProcessInstance> instances = new ArrayList<>();
        for (int i = 0; i < businessKeys.size(); i++) {
            try {
                ProcessInstance instance = runtimeService.startProcessInstanceByKey(
                        processDefinitionKey, businessKeys.get(i), variablesList.get(i));
                instances.add(instance);
            } catch (Exception e) {
                log.error("启动流程实例失败：businessKey={}", businessKeys.get(i), e);
            }
        }
        
        log.debug("批量启动流程实例完成，成功：{}/{}", instances.size(), businessKeys.size());
        return instances;
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void deleteProcessInstance(String processInstanceId, String deleteReason) {
        log.debug("删除流程实例：processInstanceId={}, deleteReason={}", 
                  processInstanceId, deleteReason);
        
        // 清除流程变量缓存
        cacheService.evictProcessVariableCache(processInstanceId);
        
        runtimeService.deleteProcessInstance(processInstanceId, deleteReason);
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void batchDeleteProcessInstances(List<String> processInstanceIds, String deleteReason) {
        log.debug("批量删除流程实例，数量：{}", processInstanceIds.size());
        
        processInstanceIds.forEach(instanceId -> {
            try {
                cacheService.evictProcessVariableCache(instanceId);
                runtimeService.deleteProcessInstance(instanceId, deleteReason);
            } catch (Exception e) {
                log.error("删除流程实例失败：processInstanceId={}", instanceId, e);
            }
        });
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void suspendProcessInstance(String processInstanceId) {
        log.debug("挂起流程实例：processInstanceId={}", processInstanceId);
        runtimeService.suspendProcessInstanceById(processInstanceId);
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void activateProcessInstance(String processInstanceId) {
        log.debug("激活流程实例：processInstanceId={}", processInstanceId);
        runtimeService.activateProcessInstanceById(processInstanceId);
    }
    
    // ========== 任务管理 ==========
    
    @Override
    @Cacheable(value = "userTasks", key = "#userId + '_' + #page + '_' + #size")
    public List<TaskInfo> getUserTasks(String userId, int page, int size) {
        log.debug("查询用户待办任务：userId={}, page={}, size={}", userId, page, size);
        
        List<Task> tasks = taskService.createTaskQuery()
                .taskAssignee(userId)
                .orderByTaskCreateTime()
                .desc()
                .listPage((page - 1) * size, size);
        
        return tasks.stream()
                .map(this::convertToTaskInfo)
                .collect(Collectors.toList());
    }
    
    @Override
    public List<TaskInfo> getUserTasksOptimized(String userId, List<String> candidateGroups) {
        log.debug("查询用户待办任务（优化版）：userId={}", userId);
        
        // 并行查询个人任务和候选组任务
        List<Task> tasks = new ArrayList<>();
        
        // 查询个人任务
        if (userId != null && !userId.isEmpty()) {
            List<Task> assignedTasks = taskService.createTaskQuery()
                    .taskAssignee(userId)
                    .orderByTaskCreateTime()
                    .desc()
                    .list();
            tasks.addAll(assignedTasks);
        }
        
        // 查询候选组任务（使用缓存）
        if (candidateGroups != null && !candidateGroups.isEmpty()) {
            for (String group : candidateGroups) {
                List<Task> candidateTasks = cacheService.getCandidateGroupTasks(group);
                if (candidateTasks == null || candidateTasks.isEmpty()) {
                    candidateTasks = taskService.createTaskQuery()
                            .taskCandidateGroup(group)
                            .orderByTaskCreateTime()
                            .desc()
                            .list();
                    cacheService.cacheCandidateGroupTasks(group, candidateTasks);
                }
                tasks.addAll(candidateTasks);
            }
        }
        
        // 去重
        Map<String, TaskInfo> uniqueTasks = new LinkedHashMap<>();
        tasks.forEach(task -> {
            uniqueTasks.put(task.getId(), convertToTaskInfo(task));
        });
        
        return new ArrayList<>(uniqueTasks.values());
    }
    
    @Override
    public List<HistoryInfo> getUserCompletedTasks(String userId, int page, int size) {
        log.debug("查询用户已办任务：userId={}, page={}, size={}", userId, page, size);
        
        List<HistoricTaskInstance> historicTasks = historyService.createHistoricTaskInstanceQuery()
                .taskAssignee(userId)
                .finished()
                .orderByHistoricTaskInstanceEndTime()
                .desc()
                .listPage((page - 1) * size, size);
        
        return historicTasks.stream()
                .map(this::convertToHistoryInfo)
                .collect(Collectors.toList());
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void completeTask(String taskId, Map<String, Object> variables) {
        log.debug("完成任务：taskId={}", taskId);
        
        Task task = taskService.createTaskQuery().taskId(taskId).singleResult();
        if (task == null) {
            throw new RuntimeException("任务不存在或已完成");
        }
        
        // 设置流程变量
        if (variables != null && !variables.isEmpty()) {
            taskService.setVariables(taskId, variables);
        }
        
        // 完成任务
        taskService.complete(taskId);
        
        // 清除相关缓存
        clearTaskCache(task.getProcessInstanceId(), task.getAssignee());
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void batchCompleteTasks(List<String> taskIds, Map<String, Object> variables) {
        log.debug("批量完成任务，数量：{}", taskIds.size());
        
        taskIds.forEach(taskId -> {
            try {
                completeTask(taskId, variables);
            } catch (Exception e) {
                log.error("完成任务失败：taskId={}", taskId, e);
            }
        });
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void claimTask(String taskId, String userId) {
        log.debug("认领任务：taskId={}, userId={}", taskId, userId);
        
        taskService.claim(taskId, userId);
        
        // 清除相关缓存
        clearTaskCache(null, userId);
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void transferTask(String taskId, String targetUserId) {
        log.debug("转办任务：taskId={}, targetUserId={}", taskId, targetUserId);
        
        Task task = taskService.createTaskQuery().taskId(taskId).singleResult();
        String oldAssignee = task.getAssignee();
        
        taskService.setAssignee(taskId, targetUserId);
        
        // 清除相关缓存
        clearTaskCache(task.getProcessInstanceId(), oldAssignee);
        clearTaskCache(task.getProcessInstanceId(), targetUserId);
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void delegateTask(String taskId, String targetUserId) {
        log.debug("委派任务：taskId={}, targetUserId={}", taskId, targetUserId);
        
        taskService.delegateTask(taskId, targetUserId);
        
        // 清除相关缓存
        clearTaskCache(null, taskService.createTaskQuery().taskId(taskId).singleResult().getAssignee());
    }
    
    // ========== 流程变量管理 ==========
    
    @Override
    @Cacheable(value = "processVariable", key = "#processInstanceId + '_' + #variableName")
    public Object getVariable(String processInstanceId, String variableName) {
        log.debug("获取流程变量：processInstanceId={}, variableName={}", 
                  processInstanceId, variableName);
        
        return runtimeService.getVariable(processInstanceId, variableName);
    }
    
    @Override
    public Map<String, Object> getVariables(String processInstanceId, List<String> variableNames) {
        log.debug("批量获取流程变量：processInstanceId={}, 变量数量={}", 
                  processInstanceId, variableNames.size());
        
        // 尝试从缓存批量获取
        Map<String, Object> cachedVariables = new HashMap<>();
        List<String> uncachedNames = new ArrayList<>();
        
        for (String name : variableNames) {
            Object value = cacheService.getProcessVariable(processInstanceId, name);
            if (value != null) {
                cachedVariables.put(name, value);
            } else {
                uncachedNames.add(name);
            }
        }
        
        // 从数据库获取未缓存的变量
        if (!uncachedNames.isEmpty()) {
            Map<String, Object> dbVariables = runtimeService.getVariables(
                    processInstanceId, uncachedNames);
            cachedVariables.putAll(dbVariables);
            
            // 缓存新获取的变量
            dbVariables.forEach((name, value) -> {
                cacheService.cacheProcessVariable(processInstanceId, name, value);
            });
        }
        
        return cachedVariables;
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void setVariables(String processInstanceId, Map<String, Object> variables) {
        log.debug("设置流程变量：processInstanceId={}, 变量数量={}", 
                  processInstanceId, variables.size());
        
        if (variables != null && !variables.isEmpty()) {
            runtimeService.setVariables(processInstanceId, variables);
            
            // 更新缓存
            variables.forEach((name, value) -> {
                cacheService.cacheProcessVariable(processInstanceId, name, value);
            });
        }
    }
    
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void batchSetVariables(List<String> processInstanceIds, String variableName, Object variableValue) {
        log.debug("批量设置流程变量：流程实例数量={}", processInstanceIds.size());
        
        processInstanceIds.forEach(instanceId -> {
            try {
                runtimeService.setVariable(instanceId, variableName, variableValue);
                cacheService.cacheProcessVariable(instanceId, variableName, variableValue);
            } catch (Exception e) {
                log.error("设置流程变量失败：processInstanceId={}", instanceId, e);
            }
        });
    }
    
    // ========== 流程历史管理 ==========
    
    @Override
    public List<HistoryInfo> getProcessHistory(String processInstanceId, int page, int size) {
        log.debug("查询流程历史：processInstanceId={}, page={}, size={}", 
                  processInstanceId, page, size);
        
        List<HistoricTaskInstance> historicTasks = historyService.createHistoricTaskInstanceQuery()
                .processInstanceId(processInstanceId)
                .orderByHistoricTaskInstanceStartTime()
                .asc()
                .listPage((page - 1) * size, size);
        
        return historicTasks.stream()
                .map(this::convertToHistoryInfo)
                .collect(Collectors.toList());
    }
    
    @Override
    public List<Map<String, Object>> getVariableHistory(String processInstanceId) {
        log.debug("查询流程变量历史：processInstanceId={}", processInstanceId);
        
        return historyService.createHistoricVariableInstanceQuery()
                .processInstanceId(processInstanceId)
                .orderByVariableName()
                .asc()
                .list()
                .stream()
                .map(var -> {
                    Map<String, Object> map = new HashMap<>();
                    map.put("variableName", var.getVariableName());
                    map.put("variableType", var.getVariableTypeName());
                    map.put("value", var.getValue());
                    map.put("createTime", var.getCreateTime());
                    return map;
                })
                .collect(Collectors.toList());
    }
    
    // ========== 辅助方法 ==========
    
    /**
     * 转换为TaskInfo
     */
    private TaskInfo convertToTaskInfo(Task task) {
        TaskInfo taskInfo = new TaskInfo();
        taskInfo.setTaskId(task.getId());
        taskInfo.setTaskName(task.getName());
        taskInfo.setTaskDescription(task.getDescription());
        taskInfo.setAssignee(task.getAssignee());
        taskInfo.setCreateTime(task.getCreateTime());
        taskInfo.setDueDate(task.getDueDate());
        taskInfo.setPriority(task.getPriority());
        taskInfo.setProcessInstanceId(task.getProcessInstanceId());
        taskInfo.setProcessDefinitionId(task.getProcessDefinitionId());
        taskInfo.setBusinessKey(task.getProcessInstanceId());
        taskInfo.setTaskKey(task.getTaskDefinitionKey());
        return taskInfo;
    }
    
    /**
     * 转换为HistoryInfo
     */
    private HistoryInfo convertToHistoryInfo(HistoricTaskInstance historicTask) {
        HistoryInfo historyInfo = new HistoryInfo();
        historyInfo.setTaskId(historicTask.getId());
        historyInfo.setTaskName(historicTask.getName());
        historyInfo.setTaskDescription(historicTask.getDescription());
        historyInfo.setAssignee(historicTask.getAssignee());
        historyInfo.setStartTime(historicTask.getCreateTime());
        historyInfo.setEndTime(historicTask.getEndTime());
        historyInfo.setDuration(historicTask.getDurationInMillis());
        historyInfo.setProcessInstanceId(historicTask.getProcessInstanceId());
        historyInfo.setProcessDefinitionId(historicTask.getProcessDefinitionId());
        historyInfo.setDeleteReason(historicTask.getDeleteReason());
        historyInfo.setTaskKey(historicTask.getTaskDefinitionKey());
        return historyInfo;
    }
    
    /**
     * 清除任务缓存
     */
    private void clearTaskCache(String processInstanceId, String userId) {
        try {
            if (userId != null) {
                // 清除用户任务缓存（使用通配符）
                cacheService.evictUserTasksCache(userId);
            }
        } catch (Exception e) {
            log.warn("清除任务缓存失败", e);
        }
    }
}

package com.example.activiti.config;

import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 异步配置类
 * 配置异步任务线程池，提升通知性能，避免阻塞主流程
 */
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
    
    /**
     * 配置异步任务线程池
     */
    @Override
    @Bean(name = "asyncExecutor")
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        
        // 核心线程数
        executor.setCorePoolSize(10);
        
        // 最大线程数
        executor.setMaxPoolSize(50);
        
        // 队列容量
        executor.setQueueCapacity(1000);
        
        // 线程空闲时间
        executor.setKeepAliveSeconds(60);
        
        // 线程名前缀
        executor.setThreadNamePrefix("async-notify-");
        
        // 拒绝策略：由调用线程执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        
        // 等待任务完成后关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        
        // 等待时间
        executor.setAwaitTerminationSeconds(60);
        
        // 允许核心线程超时
        executor.setAllowCoreThreadTimeOut(true);
        
        executor.initialize();
        return executor;
    }
    
    /**
     * 异步异常处理器
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (throwable, method, params) -> {
            // 记录异步任务异常，避免影响主流程
            System.err.println("异步任务执行异常：" + throwable.getMessage());
            throwable.printStackTrace();
        };
    }
}

package com.example.activiti.config;

import com.github.benmanes.caffeine.cache.Caffeine;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.cache.caffeine.CaffeineCacheManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.TimeUnit;

/**
 * 缓存配置类
 * 使用Caffeine高性能缓存，减少数据库查询和重复计算
 */
@Configuration
@EnableCaching
public class CacheConfig {
    
    /**
     * 配置Caffeine缓存管理器
     */
    @Bean
    public CacheManager cacheManager() {
        CaffeineCacheManager cacheManager = new CaffeineCacheManager();
        
        // 配置缓存策略
        cacheManager.setCaffeine(Caffeine.newBuilder()
                // 初始容量
                .initialCapacity(100)
                // 最大容量
                .maximumSize(5000)
                // 写入后过期时间
                .expireAfterWrite(30, TimeUnit.MINUTES)
                // 过期后自动刷新
                .refreshAfterWrite(20, TimeUnit.MINUTES)
                // 启用统计
                .recordStats()
        );
        
        // 缓存名称列表
        cacheManager.setCacheNames(
                "processDefinition",  // 流程定义缓存
                "processVariable",    // 流程变量缓存
                "userTasks",          // 用户任务缓存
                "userInfo",           // 用户信息缓存
                "emailTemplate",      // 邮件模板缓存
                "candidateGroupTasks" // 候选组任务缓存
        );
        
        return cacheManager;
    }
}

package com.example.activiti.async;

import com.example.activiti.entity.NotificationRecord;
import com.example.activiti.mapper.NotificationRecordMapper;
import com.example.activiti.service.CacheService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.mail.SimpleMailMessage;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.mail.javamail.MimeMessageHelper;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.thymeleaf.TemplateEngine;
import org.thymeleaf.context.Context;

import javax.mail.internet.MimeMessage;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 消息通知异步处理器
 * 核心性能优化：使用异步处理、批量发送、失败重试，确保通知不阻塞主流程
 */
@Slf4j
@Component
public class NotificationAsyncProcessor {
    
    @Autowired
    private JavaMailSender mailSender;
    
    @Autowired
    private TemplateEngine templateEngine;
    
    @Autowired
    private NotificationRecordMapper notificationRecordMapper;
    
    @Autowired
    private CacheService cacheService;
    
    @Value("${spring.mail.from}")
    private String fromEmail;
    
    @Value("${notification.batch-size:50}")
    private int batchSize;
    
    @Value("${notification.retry-count:3}")
    private int retryCount;
    
    @Value("${notification.retry-interval:5000}")
    private long retryInterval;
    
    // 批量发送队列
    private final BlockingQueue<NotificationRecord> sendQueue = new LinkedBlockingQueue<>(10000);
    
    /**
     * 异步发送通知（单个）
     * 关键性能优化：使用@Async注解，立即返回，不阻塞主流程
     */
    @Async("asyncExecutor")
    public void sendNotificationAsync(NotificationRecord record) {
        log.debug("异步发送通知：id={}, type={}", record.getId(), record.getNotificationType());
        
        try {
            // 尝试从缓存获取用户信息
            String recipientEmail = cacheService.getUserEmail(record.getRecipientId());
            if (recipientEmail == null) {
                recipientEmail = record.getRecipientEmail();
            }
            
            if (recipientEmail == null || recipientEmail.isEmpty()) {
                log.warn("接收人邮箱为空，跳过发送：id={}", record.getId());
                markAsFailed(record, "接收人邮箱为空");
                return;
            }
            
            // 更新状态为发送中
            updateNotificationStatus(record.getId(), NotificationRecord.NotificationStatus.PENDING.name());
            
            // 发送邮件
            boolean success = sendEmail(record, recipientEmail);
            
            if (success) {
                // 更新状态为已发送
                markAsSent(record);
                log.debug("通知发送成功：id={}", record.getId());
            } else {
                // 重试机制
                retrySend(record, recipientEmail);
            }
            
        } catch (Exception e) {
            log.error("异步发送通知失败：id={}", record.getId(), e);
            markAsFailed(record, e.getMessage());
        }
    }
    
    /**
     * 批量异步发送通知
     * 关键性能优化：批量发送，减少IO操作
     */
    @Async("asyncExecutor")
    public void batchSendNotificationsAsync(List<NotificationRecord> records) {
        log.debug("批量异步发送通知，数量：{}", records.size());
        
        int successCount = 0;
        int failCount = 0;
        
        for (NotificationRecord record : records) {
            try {
                sendNotificationAsync(record);
                successCount++;
            } catch (Exception e) {
                log.error("批量发送失败：id={}", record.getId(), e);
                failCount++;
            }
        }
        
        log.debug("批量发送完成，成功：{}，失败：{}", successCount, failCount);
    }
    
    /**
     * 重试发送通知
     */
    private void retrySend(NotificationRecord record, String email) {
        int currentRetry = record.getRetryCount() == null ? 0 : record.getRetryCount();
        
        if (currentRetry >= retryCount) {
            log.warn("达到最大重试次数，放弃发送：id={}", record.getId());
            markAsFailed(record, "达到最大重试次数");
            return;
        }
        
        // 延迟重试
        try {
            TimeUnit.MILLISECONDS.sleep(retryInterval);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        // 更新重试次数
        currentRetry++;
        record.setRetryCount(currentRetry);
        notificationRecordMapper.updateById(record);
        
        // 重新发送
        try {
            boolean success = sendEmail(record, email);
            if (success) {
                markAsSent(record);
                log.debug("重试发送成功：id={}, 重试次数={}", record.getId(), currentRetry);
            } else {
                retrySend(record, email);
            }
        } catch (Exception e) {
            log.error("重试发送失败：id={}, 重试次数={}", record.getId(), currentRetry, e);
            retrySend(record, email);
        }
    }
    
    /**
     * 发送邮件
     */
    private boolean sendEmail(NotificationRecord record, String toEmail) {
        try {
            MimeMessage message = mailSender.createMimeMessage();
            MimeMessageHelper helper = new MimeMessageHelper(message, true, "UTF-8");
            
            helper.setFrom(fromEmail);
            helper.setTo(toEmail);
            helper.setSubject(record.getNotificationTitle());
            
            // 根据通知类型选择模板
            String templateName = getTemplateName(record.getNotificationType());
            
            // 使用Thymeleaf渲染模板
            Context context = new Context();
            context.setVariable("record", record);
            context.setVariable("title", record.getNotificationTitle());
            context.setVariable("content", record.getNotificationContent());
            
            String htmlContent = templateEngine.process(templateName, context);
            helper.setText(htmlContent, true);
            
            mailSender.send(message);
            return true;
            
        } catch (Exception e) {
            log.error("发送邮件失败：toEmail={}", toEmail, e);
            return false;
        }
    }
    
    /**
     * 获取模板名称
     */
    private String getTemplateName(String notificationType) {
        switch (notificationType) {
            case "PROCESS_START":
                return "email/process-started";
            case "TASK_CREATE":
                return "email/task-created";
            case "TASK_COMPLETE":
                return "email/task-completed";
            case "PROCESS_END":
                return "email/process-ended";
            default:
                return "email/default";
        }
    }
    
    /**
     * 标记为已发送
     */
    private void markAsSent(NotificationRecord record) {
        record.setNotificationStatus(NotificationRecord.NotificationStatus.SENT.name());
        record.setSendTime(new java.util.Date());
        notificationRecordMapper.updateById(record);
    }
    
    /**
     * 标记为失败
     */
    private void markAsFailed(NotificationRecord record, String errorMessage) {
        record.setNotificationStatus(NotificationRecord.NotificationStatus.FAILED.name());
        record.setErrorMessage(errorMessage);
        notificationRecordMapper.updateById(record);
    }
    
    /**
     * 更新通知状态
     */
    private void updateNotificationStatus(Long id, String status) {
        NotificationRecord record = new NotificationRecord();
        record.setId(id);
        record.setNotificationStatus(status);
        notificationRecordMapper.updateById(record);
    }
}

package com.example.activiti.listener;

import com.example.activiti.async.NotificationAsyncProcessor;
import com.example.activiti.entity.NotificationRecord;
import com.example.activiti.mapper.NotificationRecordMapper;
import com.example.activiti.service.CacheService;
import lombok.extern.slf4j.Slf4j;
import org.activiti.engine.delegate.DelegateExecution;
import org.activiti.engine.delegate.ExecutionListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * 流程启动监听器（异步优化版）
 * 关键性能优化：监听器立即返回，通知异步发送，不阻塞流程
 */
@Slf4j
@Component("processStartListener")
public class ProcessStartListener implements ExecutionListener {
    
    @Autowired
    private NotificationRecordMapper notificationRecordMapper;
    
    @Autowired
    private NotificationAsyncProcessor notificationAsyncProcessor;
    
    @Autowired
    private CacheService cacheService;
    
    @Override
    public void notify(DelegateExecution execution) {
        long startTime = System.currentTimeMillis();
        log.debug("流程启动监听器触发：processInstanceId={}", execution.getProcessInstanceId());
        
        try {
            String processInstanceId = execution.getProcessInstanceId();
            String processDefinitionId = execution.getProcessDefinitionId();
            
            // 从流程变量中获取发起人信息
            String initiatorId = getVariable(execution, "initiatorId", "system");
            String initiatorName = getVariable(execution, "initiatorName", "系统");
            String initiatorEmail = getVariable(execution, "initiatorEmail", null);
            
            // 从缓存获取用户邮箱
            if (initiatorEmail == null) {
                initiatorEmail = cacheService.getUserEmail(initiatorId);
            }
            
            // 创建通知记录
            NotificationRecord record = new NotificationRecord();
            record.setNotificationType(NotificationRecord.NotificationType.PROCESS_START.name());
            record.setProcessInstanceId(processInstanceId);
            record.setProcessDefinitionId(processDefinitionId);
            record.setRecipientId(initiatorId);
            record.setRecipientName(initiatorName);
            record.setRecipientEmail(initiatorEmail);
            record.setSenderId("system");
            record.setSenderName("系统");
            record.setNotificationTitle("流程启动通知");
            record.setNotificationContent(buildContent(processInstanceId, initiatorName));
            record.setNotificationStatus(NotificationRecord.NotificationStatus.PENDING.name());
            
            // 保存到数据库
            notificationRecordMapper.insert(record);
            
            // 关键性能优化：异步发送通知，不阻塞主流程
            notificationAsyncProcessor.sendNotificationAsync(record);
            
            long endTime = System.currentTimeMillis();
            log.debug("流程启动监听器执行完成，耗时：{}ms", endTime - startTime);
            
        } catch (Exception e) {
            log.error("流程启动监听器执行失败", e);
            // 不抛出异常，避免影响流程执行
        }
    }
    
    /**
     * 获取流程变量
     */
    private String getVariable(DelegateExecution execution, String name, String defaultValue) {
        Object value = execution.getVariable(name);
        return value != null ? value.toString() : defaultValue;
    }
    
    /**
     * 构建通知内容
     */
    private String buildContent(String processInstanceId, String initiatorName) {
        StringBuilder content = new StringBuilder();
        content.append("尊敬的 ").append(initiatorName).append("：\n\n");
        content.append("您发起的流程已成功启动。\n");
        content.append("流程实例ID：").append(processInstanceId).append("\n");
        content.append("启动时间：").append(new java.util.Date()).append("\n");
        content.append("\n请登录系统查看详情。");
        return content.toString();
    }
}

package com.example.activiti.listener;

import com.example.activiti.async.NotificationAsyncProcessor;
import com.example.activiti.entity.NotificationRecord;
import com.example.activiti.mapper.NotificationRecordMapper;
import com.example.activiti.service.CacheService;
import lombok.extern.slf4j.Slf4j;
import org.activiti.engine.delegate.DelegateTask;
import org.activiti.engine.delegate.TaskListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 任务创建监听器（异步优化版）
 * 关键性能优化：监听器立即返回，通知异步发送，不阻塞任务创建
 */
@Slf4j
@Component("taskCreateListener")
public class TaskCreateListener implements TaskListener {
    
    @Autowired
    private NotificationRecordMapper notificationRecordMapper;
    
    @Autowired
    private NotificationAsyncProcessor notificationAsyncProcessor;
    
    @Autowired
    private CacheService cacheService;
    
    @Override
    public void notify(DelegateTask delegateTask) {
        long startTime = System.currentTimeMillis();
        log.debug("任务创建监听器触发：taskId={}", delegateTask.getId());
        
        try {
            String taskId = delegateTask.getId();
            String taskName = delegateTask.getName();
            String assigneeId = delegateTask.getAssignee();
            
            // 从缓存获取用户信息
            String assigneeName = cacheService.getUserName(assigneeId);
            String assigneeEmail = cacheService.getUserEmail(assigneeId);
            
            // 获取流程变量中的发起人信息
            String initiatorId = (String) delegateTask.getVariable("initiatorId");
            String initiatorName = (String) delegateTask.getVariable("initiatorName");
            
            // 创建通知记录
            NotificationRecord record = new NotificationRecord();
            record.setNotificationType(NotificationRecord.NotificationType.TASK_CREATE.name());
            record.setProcessInstanceId(delegateTask.getProcessInstanceId());
            record.setProcessDefinitionId(delegateTask.getProcessDefinitionId());
            record.setTaskId(taskId);
            record.setTaskName(taskName);
            record.setRecipientId(assigneeId);
            record.setRecipientName(assigneeName);
            record.setRecipientEmail(assigneeEmail);
            record.setSenderId(initiatorId);
            record.setSenderName(initiatorName);
            record.setNotificationTitle("待办任务通知");
            record.setNotificationContent(buildContent(taskName, initiatorName, assigneeName));
            record.setNotificationStatus(NotificationRecord.NotificationStatus.PENDING.name());
            
            // 保存到数据库
            notificationRecordMapper.insert(record);
            
            // 关键性能优化：异步发送通知，不阻塞任务创建
            notificationAsyncProcessor.sendNotificationAsync(record);
            
            long endTime = System.currentTimeMillis();
            log.debug("任务创建监听器执行完成，耗时：{}ms", endTime - startTime);
            
        } catch (Exception e) {
            log.error("任务创建监听器执行失败", e);
            // 不抛出异常，避免影响任务创建
        }
    }
    
    /**
     * 构建通知内容
     */
    private String buildContent(String taskName, String initiatorName, String assigneeName) {
        StringBuilder content = new StringBuilder();
        content.append("尊敬的 ").append(assigneeName).append("：\n\n");
        content.append("您有一个新的待办任务需要处理。\n");
        content.append("任务名称：").append(taskName).append("\n");
        content.append("提交人：").append(initiatorName).append("\n");
        content.append("创建时间：").append(new java.util.Date()).append("\n");
        content.append("\n请及时登录系统处理。");
        return content.toString();
    }
}

package com.example.activiti.listener;

import com.example.activiti.async.NotificationAsyncProcessor;
import com.example.activiti.entity.NotificationRecord;
import com.example.activiti.mapper.NotificationRecordMapper;
import com.example.activiti.service.CacheService;
import lombok.extern.slf4j.Slf4j;
import org.activiti.engine.delegate.DelegateTask;
import org.activiti.engine.delegate.TaskListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 任务完成监听器（异步优化版）
 * 关键性能优化：监听器立即返回，通知异步发送，不阻塞任务完成
 */
@Slf4j
@Component("taskCompleteListener")
public class TaskCompleteListener implements TaskListener {
    
    @Autowired
    private NotificationRecordMapper notificationRecordMapper;
    
    @Autowired
    private NotificationAsyncProcessor notificationAsyncProcessor;
    
    @Autowired
    private CacheService cacheService;
    
    @Override
    public void notify(DelegateTask delegateTask) {
        long startTime = System.currentTimeMillis();
        log.debug("任务完成监听器触发：taskId={}", delegateTask.getId());
        
        try {
            String taskId = delegateTask.getId();
            String taskName = delegateTask.getName();
            String assigneeId = delegateTask.getAssignee();
            String assigneeName = cacheService.getUserName(assigneeId);
            
            // 获取审批结果
            boolean approved = delegateTask.getVariable("approved") != null 
                    ? Boolean.parseBoolean(delegateTask.getVariable("approved").toString()) : true;
            String comment = delegateTask.getVariable("comment") != null 
                    ? delegateTask.getVariable("comment").toString() : "";
            
            // 获取发起人信息
            String initiatorId = (String) delegateTask.getVariable("initiatorId");
            String initiatorName = (String) delegateTask.getVariable("initiatorName");
            String initiatorEmail = cacheService.getUserEmail(initiatorId);
            
            // 创建通知记录
            NotificationRecord record = new NotificationRecord();
            record.setNotificationType(NotificationRecord.NotificationType.TASK_COMPLETE.name());
            record.setProcessInstanceId(delegateTask.getProcessInstanceId());
            record.setProcessDefinitionId(delegateTask.getProcessDefinitionId());
            record.setTaskId(taskId);
            record.setTaskName(taskName);
            record.setRecipientId(initiatorId);
            record.setRecipientName(initiatorName);
            record.setRecipientEmail(initiatorEmail);
            record.setSenderId(assigneeId);
            record.setSenderName(assigneeName);
            record.setNotificationTitle(approved ? "任务已通过" : "任务已拒绝");
            record.setNotificationContent(buildContent(taskName, assigneeName, approved, comment));
            record.setNotificationStatus(NotificationRecord.NotificationStatus.PENDING.name());
            
            // 保存到数据库
            notificationRecordMapper.insert(record);
            
            // 关键性能优化：异步发送通知，不阻塞任务完成
            notificationAsyncProcessor.sendNotificationAsync(record);
            
            long endTime = System.currentTimeMillis();
            log.debug("任务完成监听器执行完成，耗时：{}ms", endTime - startTime);
            
        } catch (Exception e) {
            log.error("任务完成监听器执行失败", e);
            // 不抛出异常，避免影响任务完成
        }
    }
    
    /**
     * 构建通知内容
     */
    private String buildContent(String taskName, String assigneeName, boolean approved, String comment) {
        StringBuilder content = new StringBuilder();
        content.append("您好：\n\n");
        content.append("任务【").append(taskName).append("】已");
        content.append(approved ? "通过" : "拒绝").append("。\n");
        content.append("审批人：").append(assigneeName).append("\n");
        if (comment != null && !comment.isEmpty()) {
            content.append("审批意见：").append(comment).append("\n");
        }
        content.append("\n处理时间：").append(new java.util.Date());
        return content.toString();
    }
}

activiti:
  process-definition-cache-limit: 500      # 增加缓存数量
  enable-process-definition-info-cache: true  # 启用流程定义信息缓存

// 流程定义缓存
@Cacheable(value = "processDefinition", key = "#processDefinitionKey")
public ProcessDefinition getProcessDefinition(String processDefinitionKey) {
    // 从数据库加载
}

// 用户信息缓存
@Cacheable(value = "userInfo", key = "#userId")
public UserInfo getUserInfo(String userId) {
    // 从数据库加载
}

// 流程变量缓存
@Cacheable(value = "processVariable", key = "#processInstanceId + '_' + #variableName")
public Object getVariable(String processInstanceId, String variableName) {
    // 从数据库加载
}

activiti:
  history-level: activity  # 仅记录活动数据，不存储详细变量历史

@Async("asyncExecutor")
public void sendNotificationAsync(NotificationRecord record) {
    // 异步发送邮件
}

// 批量完成任务
public void batchCompleteTasks(List<String> taskIds, Map<String, Object> variables) {
    taskIds.forEach(taskId -> {
        taskService.complete(taskId, variables);
    });
}

// 批量查询
List<Task> tasks = taskService.createTaskQuery()
        .taskAssignee(userId)
        .listPage(offset, limit);

public List<TaskInfo> getUserTasksOptimized(String userId, List<String> candidateGroups) {
    // 并行查询
    CompletableFuture<List<Task>> assignedTasks = CompletableFuture.supplyAsync(() -> 
            taskService.createTaskQuery().taskAssignee(userId).list());
    
    CompletableFuture<List<Task>> candidateTasks = CompletableFuture.supplyAsync(() -> 
            taskService.createTaskQuery().taskCandidateGroup("deptManager").list());
    
    // 合并结果
    return CompletableFuture.allOf(assignedTasks, candidateTasks)
            .thenApply(v -> mergeResults(assignedTasks.get(), candidateTasks.get()))
            .get();
}

spring:
  datasource:
    druid:
      initial-size: 10
      min-idle: 10
      max-active: 100
      max-wait: 60000
      test-while-idle: true
      test-on-borrow: false
      test-on-return: false
      time-between-eviction-runs-millis: 60000
      min-evictable-idle-time-millis: 300000

-- 任务表索引
CREATE INDEX idx_act_ru_task_assignee ON ACT_RU_TASK(ASSIGNEE_);
CREATE INDEX idx_act_ru_task_proc_inst ON ACT_RU_TASK(PROC_INST_ID_);
CREATE INDEX idx_act_ru_task_create_time ON ACT_RU_TASK(CREATE_TIME_);
-- 变量表索引
CREATE INDEX idx_act_ru_var_proc_inst ON ACT_RU_VARIABLE(PROC_INST_ID_);
CREATE INDEX idx_act_ru_var_name ON ACT_RU_VARIABLE(NAME_);

List<Task> tasks = taskService.createTaskQuery()
        .taskAssignee(userId)
        .orderByTaskCreateTime()
        .desc()
        .listPage(page * size, size);

java -Xms2g -Xmx2g -Xmn1g -XX:MetaspaceSize=256m -XX:MaxMetaspaceSize=512m

java -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:G1HeapRegionSize=16m

<?xml version="1.0" encoding="UTF-8"?>
<jmeterTestPlan>
  <hashTree>
    <TestPlan>
      <elementProp name="TestPlan.user_defined_variables" elementType="Arguments"/>
    </TestPlan>
    <hashTree>
      <ThreadGroup>
        <stringProp name="ThreadGroup.num_threads">100</stringProp>
        <stringProp name="ThreadGroup.ramp_time">10</stringProp>
        <stringProp name="ThreadGroup.duration">300</stringProp>
      </ThreadGroup>
    </hashTree>
  </hashTree>
</jmeterTestPlan>