搜索到
86
篇与
的结果
-
-
实施DevOps与监控 实施DevOps与监控 完成一套简单从环境安装、软件开发到应用发布的简单生命周期可持续的软件、应用管理流程。 背景 环境部署：项目实施每次部署环境都需要从头安装服务端软件，需要提一套简单流程让安装软件不需要每次都重复该项工作，并可复用，可迭代，多版本。 可持续软件发布：项目实施软件开发、测试、发布没有统一的标准、流程、规范。需要提供一套标准、规范的devops流程。 软件监控：项目实施发布后，性能、异常没有很好的监控，需要提供一套完整的监控体系进行监控和异常报警。 性能优化：提供异常、性能优化wiki供项目实施参考。 项目实施的环境安装 周期：项目实施的环境安装是一次性的。一般在项目初期的开发环境安装和测试环境安装。项目即将发布时环境的安装。 差异性：一般主要是实施单位的差异，每个公的环境不用、软件标准、服务器配置不同。 重复性：主要体现在多个实施单位、很多中间件、服务器环境是相同的。 使用Ansible进行环境的安装，该开源软件提供web ui界面帮助用户简单、流程化的完成环境的安装。 Ansible Awx的详细介绍参考Ansible Awx的介绍文档。基本的流程如下： 1. 脚本维护与版本控制 实施团队或开发团队根据需求更新 Ansible 脚本。 更改会通过 Pull Request 或合适的代码审查流程推送到 Git 仓库。 使用 Git 标签或分支管理不同版本的 Ansible 脚本，以适应不同版本的软件或不同客户的配置需求。 2. AWX 配置与准备 系统管理员或 DevOps 团队为实施团队配置 AWX，包括用户权限、团队、项目和作业模板。 在 AWX 中设置相应的 Git 仓库作为项目来源，确保 AWX 可以访问并同步该仓库。 3. 客户环境准备 实施团队与客户沟通，收集和配置目标服务器的信息。 在 AWX 中设置相应的库存，将客户的服务器信息录入。 如果需要，还可以为特定的客户或服务器组设置变量。 4. 软件部署 实施团队在 AWX 中选择合适的作业模板，并针对特定的库存执行它，以在客户的服务器上部署软件。 可以在 AWX 的 UI 中实时监控部署过程。 如有需要，可以调整 Ansible 脚本或 AWX 变量，然后重新运行作业模板。 5. 验证与确认 实施团队和/或客户验证软件是否已正确安装并按预期运行。 如有问题，返回到步骤 1 或步骤 4 进行调整。 6. 文档和培训 为客户提供任何相关的文档，说明如何使用或维护新安装的软件。 如有需要，为客户提供相关培训。 7. 支持与维护 实施团队或支持团队提供后续的技术支持，帮助客户解决可能出现的任何问题。 当软件有更新或需要变更配置时，回到步骤 1 开始新的部署周期。 这个流程提供了从脚本维护到客户软件部署的完整视图，以及后续的支持。当然，每个公司的具体需求和流程可能会有所不同，所以你可能需要根据你的实际情况进行调整。 sequenceDiagram participant 脚本开发者 participant 系统管理员 participant 实施团队 participant 客户 participant 支持团队 participant 公司 脚本开发者->>系统管理员: 推送更新的 Ansible 脚本到 Git 系统管理员->>系统管理员: 使 AWX 与 Git 仓库同步 公司->>实施团队: 提供 AWX 离线版本 客户->>实施团队: 提供目标服务器详情 实施团队->>系统管理员: 请求 AWX 访问权限 系统管理员->>实施团队: 在 AWX 上授予权限 实施团队->>AWX: 在 AWX 中配置库存 实施团队->>AWX: 运行作业模板进行软件部署 AWX->>客户: 在客户的服务器上部署软件 客户->>实施团队: 提供部署反馈 实施团队->>脚本开发者: 报告脚本中的问题或需要的更改 客户->>支持团队: 报告部署后的问题或提问 支持团队->>客户: 提供解决方案和答案 可持续软件发布 周期：软件开发、测试、发布、上线、迭代。 客户差异：客户已有devops流程与没有devops流程 复杂度：是单体应用还是微服务应用，代码管理是git还是svn。 综合考虑使用jenkins作为构建工具，主要因为jenkins专注于构建，生态完整，老牌开源产品，市场用户大，且PipeLine支持图形页面设计。详细devops流程请参考Devops流程 基于普通环境的devops流程： sequenceDiagram participant 项目经理 as 项目经理 participant 开发人员 as 开发人员 participant 测试人员 participant 运维人员 as 运维人员 participant Jenkins 项目经理->>开发人员: 定义需求 开发人员->>项目经理: 讨论技术可行性 测试人员->>项目经理: 了解需求为测试做准备 开发人员->>开发人员: 本地编码 开发人员->>Jenkins: 推送代码至版本控制系统 开发人员->>开发人员: 代码审查 Jenkins->>开发人员: 触发CI流程 开发人员->>Jenkins: 关注构建与测试结果 Jenkins->>开发人员: 执行静态代码检查 & 单元测试 Jenkins->>测试人员: 报告代码覆盖率 测试人员->>Jenkins: 使用JMeter进行性能测试 开发人员->>Jenkins: 基于性能结果优化代码 运维人员->>Jenkins: 配置预生产环境 测试人员->>Jenkins: 在预生产环境执行集成测试 运维人员->>Jenkins: 部署至生产环境 测试人员->>运维人员: 验证生产环境部署 运维人员->>Jenkins: 监控生产环境 开发人员->>测试人员: 基于生产反馈修复问题 项目经理->>所有人: 组织项目回顾 所有人->>项目经理: 分享反馈与经验教训 基于k8s部署的环境 sequenceDiagram participant 项目经理 as 项目经理 participant 开发人员 as 开发人员 participant 测试人员 participant 运维人员 as 运维人员 participant Jenkins participant K8s as Kubernetes 项目经理->>开发人员: 定义需求 开发人员->>项目经理: 讨论技术可行性 测试人员->>项目经理: 了解需求为测试做准备 开发人员->>开发人员: 本地编码 开发人员->>Jenkins: 推送代码至版本控制系统 开发人员->>开发人员: 代码审查 Jenkins->>开发人员: 触发CI流程 开发人员->>Jenkins: 关注构建与测试结果 Jenkins->>开发人员: 执行静态代码检查 & 单元测试 Jenkins->>测试人员: 报告代码覆盖率 测试人员->>Jenkins: 使用JMeter进行性能测试 开发人员->>Jenkins: 基于性能结果优化代码 Jenkins->>K8s: 使用Docker构建容器镜像 K8s->>Jenkins: 确认镜像构建成功 运维人员->>K8s: 配置预生产K8s环境 Jenkins->>K8s: 在K8s预生产环境部署应用 测试人员->>K8s: 在预生产环境执行集成测试 运维人员->>K8s: 部署应用至K8s生产环境 测试人员->>运维人员: 验证K8s生产环境部署 运维人员->>K8s: 使用K8s监控生产环境 开发人员->>测试人员: 基于生产反馈修复问题 项目经理->>所有人: 组织项目回顾 所有人->>项目经理: 分享反馈与经验教训 监控 搭建一个监控体系来监控 Kubernetes (K8s) 上运行的 Spring Boot 应用需要考虑几个方面：指标收集、日志收集、告警以及可视化。在本指南中，我们将使用 SkyWalking、Prometheus、Grafana 和 Elasticsearch (ES) 来完成这一任务。以下是如何整合这些组件的基本步骤： SkyWalking：用于分布式追踪，提供深入的性能指标和应用拓扑视图。 Prometheus：用于收集和存储指标数据。 Grafana：用于可视化 Prometheus 和 SkyWalking 的数据。 Elasticsearch (ES)：用于存储和查询日志数据。 步骤如下： 1. 搭建 SkyWalking 在 Kubernetes 集群中部署 SkyWalking。 修改 Spring Boot 应用以包含 SkyWalking Java Agent。这通常涉及将 Java Agent JAR 文件添加到 JVM 的启动参数中，并配置 Agent 连接到 SkyWalking OAP 服务器。 skyWalking可观测sql的链路执行，java线程的执行链路，http请求的执行链路、打印log日志。 2. 搭建 Prometheus 使用 prometheus-operator 在 Kubernetes 上部署 Prometheus。它允许更容易地管理 Prometheus 实例，并与 Alertmanager 和其他相关组件一起工作。 为 Spring Boot 应用配置 micrometer，使其发布 Prometheus 格式的指标。 定义 ServiceMonitor 或 PodMonitor 资源，让 Prometheus 自动发现并抓取 Spring Boot 应用的指标。 3. 搭建 Grafana 在 Kubernetes 上部署 Grafana。 配置 Grafana 使用 Prometheus 作为数据源。 导入或创建用于 Spring Boot 应用和 Kubernetes 的仪表板。 4. 集成 Elasticsearch 在 Kubernetes 上部署 Elasticsearch。 配置 Spring Boot 使用 logback 或其他日志框架将日志发送到 Elasticsearch。 可以使用 Kibana 或 Grafana 为 Elasticsearch 创建仪表板，以可视化日志数据。 5. 警告和通知 使用 Prometheus 的 Alertmanager 来定义警告规则和接收警告。 配置 Alertmanager 将警告发送到所需的通知渠道，如 Slack、Email 等。 此步骤只是概述，具体的实施可能涉及更多的配置和调整，特别是考虑到 Kubernetes 和 Spring Boot 应用的特定需求和环境。确保在生产环境部署前进行充分的测试。 更多业务链路日志可以通过自定义skywalking自定义封装 kibana是否使用可以再讨论，应该不需要复杂的日志查询
-
聊一聊规则引擎 规则引擎能力 一、规则集 规则集也叫决策集，是由一组普通规则和循环规则构成的规则集合，是使用频率最高的一种业务规则实现方式。 鲁班提供循环、逻辑（IF/ELSE）、赋值、执行、跳出规则等指令。 经典实例 新用户注册优惠： 如果用户是新注册用户，并且首次购买订单金额超过一定数额，则发送一张10%折扣的优惠券。 购物满减优惠： 如果用户购物车中的商品总价达到一定金额阈值，则自动应用满减优惠。比如，购物车金额大于100元可以享受减免10元的优惠。 特定商品优惠： 如果用户购买特定商品（如促销商品、季节性商品），则提供针对该商品额外的折扣或优惠。 生日优惠券： 在用户生日时发送一张生日优惠券，提供额外的折扣或者赠送特定商品。 用户活跃度奖励： 对于长时间未下单的用户，发送特定额度的优惠券以鼓励再次购买。 地理位置优惠： 针对特定地理位置的用户发送定向的优惠券，以促进当地的销售。 病人风险评估系统规则集示例 考虑以下条件和操作： 年龄： 30岁以下（Young）、30到50岁（Middle-aged）、50岁以上（Elderly）。 生活方式： 锻炼频率（High、Moderate、Low）、饮食习惯（Healthy、Unhealthy）。 慢性病史： 有（Yes）或无（No）。 风险评估等级： 低风险（Low）、中风险（Moderate）、高风险（High）。 规则集示例： FOR（用户集合 ） 用户。年龄 IF 年龄 = Elderly AND (锻炼频率 = Low OR 饮食习惯 = Unhealthy) THEN 风险评估等级 = 高风险 IF 年龄 = Middle-aged AND 慢性病史 = Yes THEN 风险评估等级 = 高风险 IF 年龄 = Young AND 锻炼频率 = High THEN 风险评估等级 = 低风险 IF 年龄 = Elderly AND 锻炼频率 = Moderate THEN 风险评估等级 = 中风险 在这个例子中，规则集根据病人的年龄、生活方式和慢性病史等条件制定了不同的风险评估规则。例如，对于年龄较大且锻炼频率低或饮食不健康的病人，系统会评估其为高风险群体，需要更频繁的检查或干预措施。 这种规则集的应用有助于医疗保健领域更精准地对病人进行风险评估，并根据其个体情况提供相应的医疗建议，从而改善病人的健康状况。 二、决策树 决策树又称为规则树，规则引擎中提供的另外一种构建规则的方式，它以一棵树形结构来表现规则，决策树表现业务规则更为形象，其中每个节点代表一个测试条件，每个分支代表一个测试结果，而每个叶子节点包含一个决策结果。 信用评分系统决策树示例 考虑以下特征和目标： 年龄： 25岁以下（Young）、25到40岁（Middle-aged）、40岁以上（Elderly）。 收入水平： 低收入（Low）、中等收入（Medium）、高收入（High）。 信用历史： 良好（Good）、一般（Fair）、差劣（Poor）。 是否有抵押物： 有（Yes）或无（No）。 信用申请结果： 批准（Approved）或拒绝（Rejected）。 决策树示例： IF 信用历史 = 良好 AND 收入水平 = 高收入 THEN 信用申请结果 = 批准 ELSE IF 信用历史 = 良好 AND 收入水平 = 中等收入 THEN 信用申请结果 = 批准 ELSE IF 信用历史 = 一般 AND 是否有抵押物 = 有 THEN 信用申请结果 = 批准 ELSE IF 信用历史 = 一般 AND 是否有抵押物 = 无 THEN 信用申请结果 = 拒绝 ELSE IF 信用历史 = 差劣 THEN 信用申请结果 = 拒绝 在这个例子中，决策树根据申请者的年龄、收入水平、信用历史和是否有抵押物等特征，自动判断信用申请的结果。例如，如果申请者信用历史良好且收入水平高，那么系统会批准其信用申请；而如果信用历史一般但有抵押物，同样会批准。 这种决策树的应用使得信用评分系统能够自动化、可解释地做出信用决策，提高了效率同时保持了透明度。 三、决策表 决策表是一种以表格形式表现规则的工具，它非常适用于描述处理判断条件较多，各条件又相互组合、有多种决策方案的情况，决策表提供精确而简洁描述复杂逻辑的方式。 视图 使用excel 决策表的应用场景： 业务规则管理： 用于管理和维护大量的业务规则，使得业务规则的管理更加简便和直观。 风险评估： 在金融领域，可以使用决策表来定义风险评估规则，根据客户的不同特征进行信用评级。 产品定价： 在销售领域，可以使用决策表来制定产品定价策略，根据市场条件和产品特性进行灵活的定价。 合规性检查： 在法务和合规性领域，决策表可以用于定义符合法规的业务行为，确保组织的合规性。 推荐系统： 在电商领域，可以使用决策表来定义推荐算法的规则，根据用户的行为和喜好推荐相应的产品 电子商务促销活动管理系统决策表示例 假设我们有以下条件和操作： 用户类型： 普通用户（Regular）和会员用户（Premium）。 购物车金额： 低于100元（Low）和100元及以上（High）。 商品类别： 电子产品（Electronics）、服装（Apparel）、食品（Food）。 促销策略： 折扣（Discount）、满减（Cashback）、免费赠品（Free Gift）。 决策表： 用户类型 购物车金额 商品类别 触发促销策略 操作 普通用户 低金额 电子产品 折扣 应用10%折扣 普通用户 低金额 服装 满减 满200减20 普通用户 高金额 食品 免费赠品 赠送一份免费小吃 会员用户 高金额 电子产品 满减 满500减50 会员用户 低金额 服装 折扣 应用15%折扣 会员用户 高金额 食品 免费赠品 赠送一份免费礼包 在这个例子中，决策表根据用户类型、购物车金额和商品类别等条件定义了不同的促销策略。例如，对于普通用户购买低金额的电子产品，系统会应用10%的折扣；而对于会员用户购买高金额的电子产品，系统则会提供满减优惠。 这样的决策表使得促销活动管理系统能够根据多个条件智能地选择适当的促销策略，提高销售效果和用户满意度。 四、交叉决策表 交叉决策表又叫决策矩阵，是一种特殊类型的决策表。 与普通决策表相比，交叉决策表的条件由纵向和横向两个维度决定，而普通决策表的条件只是由纵向维度决定；但在普通决策表的动作部分可以是三种类型，分别是赋值、输出和执行方式，而在交叉决策表中动作部分就是纵向和横向两个维度交叉后的单元格的值，一般来说，这种交叉后单元格的值都是赋给某个变量或参数，所以交叉决策表的动作基本就一个，那就是赋值。 实际业务场景 假设我们正在开发一个智能家居系统，通过交叉决策表来确定何时触发不同设备的自动化操作。以下是一个简化的实际应用场景： 智能家居系统交叉决策表示例 考虑以下几个条件和设备： 时间段： 白天（Daytime）和夜晚（Nighttime）。 人员在家： 有人在家（Occupied）和无人在家（Unoccupied）。 光照程度： 光线充足（Well-lit）和光线不足（Dimly-lit）。 设备： 灯光（Lights）、温控器（Thermostat）、安全摄像头（Security Camera）。 交叉决策表： 时间段 人员在家 光照程度 触发设备 操作 白天 有人 光线充足 无 保持设备关闭 白天 有人 光线不足 灯光 打开灯光 白天 无人 无 灯光、温控器 打开灯光、调整温度 夜晚 有人 光线充足 无 保持设备关闭 夜晚 有人 光线不足 灯光 打开灯光 夜晚 无人 无 安全摄像头 打开安全摄像头 在这个例子中，交叉决策表综合考虑了时间段、人员在家与否以及光照程度等多个条件。根据这些条件的不同组合，系统可以智能地触发相应的设备操作，使智能家居系统更符合用户的实际需求。例如，在夜晚、有人在家、光线不足的情况下，系统可以自动打开灯光，提高家居的舒适性和安全性。 五、评分卡 评分是对个人或机构的相关信息进行分析之后的一种数值表达，表示此人或此机构由于信用活动的拒付行为所造成损失风险的可能性，评分通常用于对个人或机构的风险管理与评估。 假设我们正在构建一个贷款评分系统，使用评分卡来评估个人信用并决定是否批准其贷款申请。以下是一个简化的实际应用场景： 贷款评分卡示例 考虑以下特征和分数分配： 年龄： 20岁以下（0分）、20到30岁（10分）、30到40岁（20分）、40岁以上（30分）。 收入水平： 低收入（0分）、中等收入（15分）、高收入（30分）。 信用历史： 良好（30分）、一般（15分）、差劣（0分）。 负债比例： 低于30%（20分）、30%到50%（10分）、50%以上（0分）。 工作年限： 不足一年（0分）、1到5年（10分）、5年以上（20分）。 评分卡示例： 总分 = 年龄分 + 收入分 + 信用历史分 + 负债比例分 + 工作年限分 IF 总分 >= 70 THEN 批准贷款 ELSE IF 50 <= 总分 < 70 THEN 需要进一步审查 ELSE 批准贷款 在这个例子中，评分卡通过对申请者的年龄、收入水平、信用历史、负债比例和工作年限等特征进行分数分配，计算出总分。然后，系统通过总分的阈值来决定是否批准贷款。例如，总分大于等于70分的申请者可以直接批准贷款，而总分在50到70之间的申请者需要进一步审查。 这种评分卡的应用使得贷款决策更加客观、可量化，有助于提高贷款审批的效率和一致性。 现有规则引擎在业务端使用评估 业务规则集适合简单IF - THEN逻辑业务和简单循环业务逻辑 决策树适合简单IF - ELSE 带分支的简单业务逻辑使用 复杂多条件场景想使用决策表或者评估卡来创建规则，更符合业务人员的使用习惯 如何使用规则引擎 sequenceDiagram participant BusinessUser as 业务人员 participant BusinessIT as 业务IT人员 participant Implementation as 公司实施人员 BusinessUser ->> BusinessIT: 请求创建具体规则 BusinessIT ->> Implementation: 使用规则引擎配置规则模板和规则包 Implementation ->> BusinessIT: 完成规则配置 BusinessIT ->> BusinessUser: 使用规则模板创建规则 BusinessUser ->> BusinessIT: 提出规则调整需求 BusinessIT ->> BusinessUser: 根据需求调整规则逻辑 BusinessUser ->> BusinessIT: 动态创建和维护规则 在整个Saas系统中，涉及到三个主要角色：业务人员、业务IT人员和公司实施人员。以下是每个角色在规则引擎使用过程中的主要职责： 公司实施人员 规则集、决策树、决策表、交叉决策表、评估卡的规则包创建： 负责使用规则引擎的界面或工具，创建和配置规则集、决策树、决策表、交叉决策表、评估卡的规则包，以便后续的规则模板和规则创建。 业务规则模板的创建： 设计和创建通用的业务规则模板，为业务IT人员提供规则创建的基础结构和规范。 业务IT人员 规则创建： 根据公司实施人员配置好的规则模板和知识包，使用规则引擎的工具创建具体的业务规则，包括规则集、决策树、决策表、交叉决策表、评估卡等。 规则逻辑调整： 负责根据业务需求对已创建的规则进行逻辑调整，确保规则符合业务规则模板和公司实施人员的要求。 规则维护： 定期检查和更新规则，确保规则引擎中的规则集和决策树等结构始终与业务要求保持一致。 业务人员 业务规则调整： 在规则引擎的用户界面中，根据业务需求直接调整已创建的规则，例如修改决策表的条件或调整评估卡的分数等。 规则逻辑优化： 根据业务实际运作的反馈，提出对规则逻辑的优化建议，协助业务IT人员进行调整。 规则动态创建和维护： 在业务变化时，通过规则引擎的界面灵活地创建和维护规则，确保系统能够快速响应业务的变化。 这样的角色划分和职责分配可以使得规则引擎的使用更具灵活性和适应性，不同层次的用户可以专注于其核心职责，共同完成规则的创建和维护。 定制化业务 部分更针对性业务模式由业务团队进行定制页面，如果需要更多定制指令由技术部开发
-
Flowable工作流引擎源码深度解析 Flowable工作流引擎源码深度解析 前言 Flowable是一个轻量级的业务流程引擎，基于BPMN 2.0规范实现，是Activiti项目的一个分支。作为Java生态中最流行的工作流引擎之一，了解其内部实现对于定制化开发和性能优化至关重要。本文将深入分析Flowable的核心源码结构和执行逻辑，帮助开发者更好地理解和使用这一强大工具。 核心架构概览 Flowable的源码主要分为以下几个核心模块： flowable-engine：核心引擎实现 flowable-bpmn-converter：BPMN模型转换器 flowable-process-validation：流程验证模块 flowable-image-generator：流程图生成模块 flowable-rest：REST API实现 其中，flowable-engine是最核心的部分，我们的分析也将主要集中在这个模块上。 ProcessEngine初始化流程 Flowable的入口是ProcessEngine接口，通常通过ProcessEngineConfiguration来创建。让我们看看其初始化过程： public class ProcessEngineConfigurationImpl extends ProcessEngineConfiguration { public ProcessEngine buildProcessEngine() { init(); ProcessEngineImpl processEngine = new ProcessEngineImpl(this); postProcessEngineInitialisation(); return processEngine; } public void init() { initCommandContextFactory(); initTransactionContextFactory(); initCommandExecutors(); initServices(); initDataSource(); initDbSchema(); initBeans(); initTransactionFactory(); // 其他初始化方法... } } 初始化过程主要包括： 初始化命令上下文工厂 初始化事务上下文工厂 初始化命令执行器 初始化各种服务 初始化数据源和数据库架构 初始化事务工厂等 这种设计遵循了良好的工厂模式和构建器模式。 命令模式的应用 Flowable大量使用了命令模式，所有对流程的操作都被封装为Command对象： public interface Command<T> { T execute(CommandContext commandContext); } 执行命令的是CommandExecutor，它主要有两个实现： CommandExecutorImpl：普通实现 TransactionCommandExecutor：带事务的实现 让我们看看CommandExecutorImpl的实现： public class CommandExecutorImpl implements CommandExecutor { protected CommandContextFactory commandContextFactory; protected TransactionContextFactory transactionContextFactory; public <T> T execute(Command<T> command) { CommandContext commandContext = commandContextFactory.createCommandContext(command); try { T result = command.execute(commandContext); commandContext.close(); return result; } catch (Exception e) { commandContext.exception(e); } finally { try { commandContext.close(); } catch (Exception e) { // 日志记录 } } return null; } } 这种设计确保了所有流程操作都在一个一致的上下文中执行，并且可以正确处理事务和异常。 流程定义加载 当我们部署一个BPMN文件时，Flowable会解析它并转换为内部模型。核心类是BpmnParser： public class BpmnParser { public BpmnParse createParse() { return new BpmnParse(this); } public BpmnParse parse(InputStream inputStream) { BpmnParse bpmnParse = createParse(); bpmnParse.sourceInputStream = inputStream; bpmnParse.execute(); return bpmnParse; } } BpmnParse类负责具体的解析逻辑： public class BpmnParse extends BpmnParseHandler { public BpmnParse execute() { try { // 解析XML文档 DocumentBuilderFactory dbf = XmlUtil.createSafeDocumentBuilderFactory(); Document document = dbf.newDocumentBuilder().parse(sourceInputStream); // 解析BPMN元素 parseRootElement(document.getDocumentElement()); // 处理流程定义 processDI(); // 完成解析 executeParse(); } catch (Exception e) { // 异常处理 } return this; } protected void parseRootElement(Element rootElement) { // 解析流程、任务、网关等元素 } } 这个过程将BPMN文件转换为ProcessDefinitionEntity对象，其中包含了流程的所有信息。 流程执行引擎 流程实例的执行由ExecutionEntity类负责： public class ExecutionEntity implements Execution, ExecutionListenerContainer { protected String id; protected ProcessDefinitionEntity processDefinition; protected String businessKey; protected String activityId; protected ExecutionEntity parent; protected List<ExecutionEntity> executions = new ArrayList<>(); // 其他属性... public void start() { CommandContext commandContext = Context.getCommandContext(); // 执行流程实例启动逻辑 // 触发事件监听器 // 执行第一个活动节点 } public void continueExecution() { ExecutionEntity execution = this; while (execution != null && execution.isActive()) { ActivityImpl activity = execution.getActivity(); if (activity != null) { // 执行当前活动节点 execution = activity.execute(execution); } else { // 已完成 execution = null; } } } } 每个流程实例都对应一个ExecutionEntity，每个并行流程也对应一个子ExecutionEntity。 任务管理 任务由TaskEntity类表示： public class TaskEntity implements Task { protected String id; protected String name; protected String description; protected String assignee; protected Date createTime; protected String executionId; protected String processInstanceId; // 其他属性... public void complete() { // 验证任务状态 // 执行任务完成逻辑 // 触发事件监听器 // 推进流程执行 } } 任务的创建、分配和完成都是通过TaskService接口实现的： public interface TaskService { Task newTask(); void saveTask(Task task); void deleteTask(String taskId); void claim(String taskId, String userId); void complete(String taskId); // 其他方法... } 事件监听机制 Flowable提供了丰富的事件监听机制，核心接口是FlowableEventListener： public interface FlowableEventListener { void onEvent(FlowableEvent event); boolean isFailOnException(); } 事件类型由FlowableEventType枚举定义，包括流程启动、任务创建、流程完成等多种类型。 事件的分发由FlowableEventDispatcher接口负责： public interface FlowableEventDispatcher { void addEventListener(FlowableEventListener listener); void addEventListener(FlowableEventListener listener, FlowableEventType... types); void removeEventListener(FlowableEventListener listener); void dispatchEvent(FlowableEvent event); } 这种设计允许我们在流程的各个阶段插入自定义逻辑。 数据持久化 Flowable使用MyBatis作为ORM框架进行数据持久化。核心接口是DbSqlSession： public class DbSqlSession implements Session { protected SqlSession sqlSession; protected DbSqlSessionFactory dbSqlSessionFactory; protected List<PersistentObject> insertedObjects = new ArrayList<>(); protected List<PersistentObject> updatedObjects = new ArrayList<>(); protected List<PersistentObject> deletedObjects = new ArrayList<>(); public void flush() { // 处理插入对象 for (PersistentObject insertedObject : insertedObjects) { String insertStatement = dbSqlSessionFactory.getInsertStatement(insertedObject); sqlSession.insert(insertStatement, insertedObject); } // 处理更新对象 for (PersistentObject updatedObject : updatedObjects) { String updateStatement = dbSqlSessionFactory.getUpdateStatement(updatedObject); sqlSession.update(updateStatement, updatedObject); } // 处理删除对象 for (PersistentObject deletedObject : deletedObjects) { String deleteStatement = dbSqlSessionFactory.getDeleteStatement(deletedObject); sqlSession.delete(deleteStatement, deletedObject); } // 清空缓存 insertedObjects.clear(); updatedObjects.clear(); deletedObjects.clear(); } } 所有实体对象的变更都会被记录在这些列表中，然后在事务提交时一次性写入数据库。 性能优化 Flowable做了许多性能优化，其中最重要的是缓存机制： public class DeploymentCache<T> { protected Map<String, T> cache = new HashMap<>(); protected int limit; protected LinkedList<String> keyList = new LinkedList<>(); public void add(String id, T obj) { if (limit > 0 && keyList.size() >= limit) { String oldestKey = keyList.removeFirst(); cache.remove(oldestKey); } cache.put(id, obj); keyList.addLast(id); } public T get(String id) { return cache.get(id); } public void remove(String id) { cache.remove(id); keyList.remove(id); } } 这种LRU缓存确保了频繁使用的流程定义可以快速获取，而不需要重复从数据库加载。 扩展点分析 Flowable提供了丰富的扩展点，允许开发者定制化流程行为： TaskListener：任务生命周期监听器 ExecutionListener：执行流程监听器 ActivityBehavior：自定义活动行为 ExpressionManager：表达式管理器 VariableType：自定义变量类型 以TaskListener为例： public interface TaskListener { String EVENTNAME_CREATE = "create"; String EVENTNAME_ASSIGNMENT = "assignment"; String EVENTNAME_COMPLETE = "complete"; String EVENTNAME_DELETE = "delete"; void notify(DelegateTask delegateTask); } 通过实现这个接口，我们可以在任务的各个生命周期阶段插入自定义逻辑。 异步执行器 Flowable 6引入了新的异步执行器，替代了旧的JobExecutor： public class DefaultAsyncExecutor implements AsyncExecutor { protected ThreadPoolExecutor threadPoolExecutor; protected RejectedExecutionHandler rejectedExecutionHandler; protected ThreadFactory threadFactory; public void start() { if (threadPoolExecutor == null) { threadPoolExecutor = new ThreadPoolExecutor( corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(queueSize), threadFactory, rejectedExecutionHandler ); } // 启动作业获取线程 startJobAcquisitionThread(); } protected void executeAsyncJob(Job job) { Runnable runnable = new ExecuteAsyncRunnable(job, this); threadPoolExecutor.execute(runnable); } } 这种设计使得Flowable可以高效地处理大量的异步任务。 我很乐意解释Flowable 6.x中如何实现动态加节点，并提供一个实用的demo。 在Flowable 6.x中，动态修改流程实例主要通过RuntimeService的API来实现，特别是通过createProcessInstanceModification方法。下面我将详细介绍实现方法并提供一个完整的示例。 package com.example.flowable; import org.flowable.engine.*; import org.flowable.engine.impl.cfg.StandaloneProcessEngineConfiguration; import org.flowable.engine.repository.Deployment; import org.flowable.engine.repository.ProcessDefinition; import org.flowable.engine.runtime.ProcessInstance; import org.flowable.task.api.Task; import java.util.HashMap; import java.util.List; import java.util.Map; /** * Flowable 6.x 动态添加节点示例 */ public class FlowableDynamicTaskDemo { public static void main(String[] args) { // 初始化流程引擎 ProcessEngine processEngine = initProcessEngine(); // 获取各种服务 RepositoryService repositoryService = processEngine.getRepositoryService(); RuntimeService runtimeService = processEngine.getRuntimeService(); TaskService taskService = processEngine.getTaskService(); // 部署流程定义 Deployment deployment = repositoryService.createDeployment() .addClasspathResource("dynamic-process.bpmn20.xml") .deploy(); ProcessDefinition processDefinition = repositoryService.createProcessDefinitionQuery() .deploymentId(deployment.getId()) .singleResult(); System.out.println("流程定义部署完成: " + processDefinition.getName()); // 启动流程实例 Map<String, Object> variables = new HashMap<>(); variables.put("applicant", "张三"); variables.put("amount", 5000); ProcessInstance processInstance = runtimeService.startProcessInstanceByKey("dynamicProcess", variables); System.out.println("流程实例启动成功，ID: " + processInstance.getId()); // 查询当前任务 Task currentTask = taskService.createTaskQuery() .processInstanceId(processInstance.getId()) .singleResult(); System.out.println("当前任务: " + currentTask.getName()); // 动态添加一个审核任务节点 System.out.println("开始动态添加节点..."); dynamicallyAddTask(runtimeService, processInstance.getId(), currentTask.getId()); // 完成当前任务 taskService.complete(currentTask.getId()); System.out.println("完成任务: " + currentTask.getName()); // 查看动态添加的任务 List<Task> tasks = taskService.createTaskQuery() .processInstanceId(processInstance.getId()) .list(); for (Task task : tasks) { System.out.println("当前活动任务: " + task.getName() + ", ID: " + task.getId()); // 完成动态添加的任务 taskService.complete(task.getId()); System.out.println("完成任务: " + task.getName()); } // 检查流程是否结束 ProcessInstance runningInstance = runtimeService.createProcessInstanceQuery() .processInstanceId(processInstance.getId()) .singleResult(); if (runningInstance == null) { System.out.println("流程实例已完成"); } else { System.out.println("流程实例仍在运行"); } } /** * 动态添加任务节点 */ private static void dynamicallyAddTask(RuntimeService runtimeService, String processInstanceId, String activityId) { runtimeService.createProcessInstanceModification(processInstanceId) .startBeforeActivity(activityId) // 在当前活动前启动 .addExecution() // 添加一个执行分支 .callActivity("dynamicTask") // 调用一个动态任务活动 .setVariable("reviewer", "李四") // 设置变量 .setVariable("dynamicTaskName", "财务经理审核") // 动态任务名称 .execute(); System.out.println("动态任务添加成功"); } /** * 初始化流程引擎 */ private static ProcessEngine initProcessEngine() { ProcessEngineConfiguration cfg = new StandaloneProcessEngineConfiguration() .setJdbcUrl("jdbc:h2:mem:flowable;DB_CLOSE_DELAY=-1") .setJdbcUsername("sa") .setJdbcPassword("") .setJdbcDriver("org.h2.Driver") .setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE); return cfg.buildProcessEngine(); } } <?xml version="1.0" encoding="UTF-8"?> <definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:flowable="http://flowable.org/bpmn" targetNamespace="http://www.flowable.org/processdef"> <process id="dynamicProcess" name="动态节点示例流程" isExecutable="true"> <startEvent id="startEvent" name="开始"/> <sequenceFlow id="flow1" sourceRef="startEvent" targetRef="firstTask"/> <userTask id="firstTask" name="部门经理审批" flowable:assignee="${applicant}"> <documentation>申请金额: ${amount}</documentation> </userTask> <sequenceFlow id="flow2" sourceRef="firstTask" targetRef="dynamicTask"/> <userTask id="dynamicTask" name="动态任务" flowable:assignee="${reviewer}"> <documentation>这是一个可以被动态添加的任务节点</documentation> </userTask> <sequenceFlow id="flow3" sourceRef="dynamicTask" targetRef="endEvent"/> <endEvent id="endEvent" name="结束"/> </process> </definitions> 在Flowable 6.x中动态添加节点主要有以下几种方法： 主要方法 流程实例修改 - 通过RuntimeService提供的API： runtimeService.createProcessInstanceModification(processInstanceId) .startBeforeActivity("activityId") // 在指定活动前启动 .execute(); 动态子流程注入 - 在运行时动态添加子流程： runtimeService.createProcessInstanceModification(processInstanceId) .addExecution() .callActivity("subProcessId") .execute(); 动态创建活动 - 直接创建和执行新活动： runtimeService.createProcessInstanceModification(processInstanceId) .startBeforeActivity("userTaskId") .setVariable("assignee", "动态分配的用户") .execute(); 使用场景 动态添加节点在以下场景特别有用： 基于业务规则动态调整审批流程 根据申请金额/内容添加额外的审核步骤 特殊情况下插入临时审核/处理节点 灵活处理流程异常情况 实现要点 确保节点ID唯一性：动态添加的节点必须有唯一的ID 维护执行链的完整性：确保流程能正确流转 变量传递：为动态节点设置必要的流程变量 权限管理：动态添加的任务需要正确分配处理人 注意事项 过度使用动态节点会使流程追踪和维护变得困难 动态节点可能不会显示在流程图上，需要额外的审计记录 升级到Flowable 7时，部分API可能会有变化 确保动态添加的节点在流程异常时能够被正确处理 这个示例展示了如何在Flowable 6.x版本中动态添加一个审核任务节点。你可以根据实际需求调整代码，例如添加多个节点、条件节点或并行节点。 结论 通过深入分析Flowable的源码，我们可以看到它采用了许多优秀的设计模式： 命令模式：封装所有流程操作 工厂模式：创建各种对象 构建器模式：配置引擎 策略模式：不同的活动行为 观察者模式：事件监听机制 这些设计使得Flowable既灵活又强大，能够适应各种复杂的业务流程需求。同时，它的性能优化策略也确保了在高负载环境下的稳定运行。 对于想要深入了解工作流引擎实现的开发者，Flowable的源码提供了一个很好的学习案例。通过理解其内部机制，我们可以更好地使用和扩展这个强大的引擎。 参考资源 Flowable GitHub仓库：https://github.com/flowable/flowable-engine Flowable官方文档：https://www.flowable.org/docs/userguide/index.html BPMN 2.0规范：https://www.omg.org/spec/BPMN/2.0/
-
分布式锁如何自动续期（Watchdog 机制解析） 分布式锁如何自动续期（Watchdog 机制解析） 在分布式系统中，分布式锁广泛应用于防止多个进程同时操作共享资源。其中，基于 Redis 的分布式锁（如 Redisson）提供了一种高效的实现方式。然而，锁的超时管理是一个关键问题：如果业务执行时间较长，锁可能会在任务完成前过期，从而导致多个客户端同时持有锁，造成数据竞争。为了解决这个问题，我们可以使用 Watchdog（看门狗）机制，实现分布式锁的自动续期。 1. 分布式锁的基本流程 以 Redis 为例，分布式锁的获取和释放流程通常如下： 客户端1 发送 SET lock_key unique_id NX PX 30000 请求，尝试获取锁。 Redis 服务器 存储 lock_key，过期时间设定为 30 秒。 客户端1 成功获取锁，开始执行任务。 如果任务在 30 秒内完成，客户端1 发送 DEL lock_key 释放锁。 其他客户端（如 客户端2）在锁释放后才能成功获取锁。 上述流程有一个关键问题：如果 客户端1 任务执行时间超过 30 秒，但还未完成，锁会被 Redis 自动删除，从而导致其他客户端获取锁，出现并发问题。 2. Watchdog 机制的作用 为了解决锁过期的问题，Redisson 等库引入了 Watchdog 机制。Watchdog 的基本原理如下： 客户端1 获取锁后，启动 Watchdog 线程，每隔 lockLeaseTime / 3 秒（例如 10 秒）检查锁是否仍然持有。 如果 客户端1 仍然活跃，则自动向 Redis 发送 PEXPIRE lock_key 30000 请求，延长锁的有效期。 当 客户端1 任务执行完成 时，主动调用 DEL lock_key 释放锁。 如果 客户端1 进程崩溃，Watchdog 不会继续续期，锁会在到期后自动释放。 3. Watchdog 续期的流程示意 根据你的流程图，分布式锁 Watchdog 续期的流程如下： 客户端1 获取锁（SET lock_key unique_id NX PX 30000）。 客户端1 启动 Watchdog 线程，间隔 10 秒检查锁状态。 Watchdog 检查 Redis 是否仍然持有 lock_key。 如果 lock_key 仍然属于 客户端1，执行 PEXPIRE lock_key 30000 延长锁的有效期。 如果 客户端1 任务完成，调用 DEL lock_key 释放锁，Watchdog 线程退出。 如果 客户端1 进程意外终止，Watchdog 失效，锁超时后自动释放。 4. Watchdog 机制的优缺点 优点 自动续期：任务执行时间不确定时，确保锁不会过早释放。 进程故障自动释放：如果进程崩溃，锁在超时后自动释放，避免死锁。 减少人为干预：无需手动计算任务时间，提高系统健壮性。 缺点 需要持久化心跳检测：Watchdog 线程需要定期与 Redis 通信，增加了 Redis 负载。 续期间隔需优化：如果 Watchdog 续期间隔过长，锁可能仍然会过期。 5. 代码示例（基于 Redisson） // 创建 Redisson 客户端 RedissonClient redisson = Redisson.create(); // 获取分布式锁 RLock lock = redisson.getLock("myLock"); boolean locked = lock.tryLock(10, 30, TimeUnit.SECONDS); if (locked) { try { // 执行业务逻辑 System.out.println("业务处理中..."); Thread.sleep(40000); // 模拟业务执行超时 } finally { // 释放锁 lock.unlock(); } } 在 Redisson 中，如果 lock.tryLock() 仅传递 租约时间 lockLeaseTime，则默认会启用 Watchdog 机制，自动续期，直到 unlock() 被调用。 6. 总结 Watchdog 机制通过定期续约，避免 Redis 锁过期导致的并发问题。 适用于执行时间不确定的任务，能够在任务完成前自动续期。 需要权衡 Watchdog 续期频率和 Redis 负载，以优化性能。 如果你的业务场景对分布式锁有较高的可靠性要求，建议使用 Redisson 内置的 Watchdog 机制，而非手动管理锁的生命周期。
