Skip to main content
版本：v1.9.0

EventMesh 工作流

业务场景​

图中你正在构建一个简单的电商订单管理系统，系统能够接收和调配新的订单，调配流程需要处理所有的订单创建，付款处理以及发货处理。

为了实现高可用和高性能，你可以使用事件驱动架构（EDA）构建微服务应用去处理商店前端，订单管理，支付处理和发货管理。你可以在云上部署整个系统。要处理高并发，你可以利用消息系统缓冲，并扩展多个微服务实例。架构类似于：

Workflow Use Case

当每个微服务都在自己的事件通道上运行时，EventMesh在执行事件编排方面发挥着至关重要的作用。

我们使用 CNCF Serverless工作流 来描述此事件工作流编排。

CNCF Serverless工作流​

CNCF Serverless工作流定义了一个厂商中立、开源和完全社区驱动的生态系统，用于定义和运行针对Serverless技术领域的基于DSL的工作流。

Serverless工作流定义了一种领域特定语言（DSL）来描述有状态和无状态的基于工作流的serverless函数和微服务编排。

详见官方github

EventMesh工作流​

我们利用Serverless工作流DSL来描述EventMesh工作流。根据其规范，工作流由一系列用于描述控制流逻辑的工作流状态组成。目前，我们仅支持与事件相关的工作流状态。请参见工作流DSL设计中支持的状态。

工作流状态可以包含通用的操作，或在工作流执行期间应调用的服务/函数。这些操作可以引用可复用的函数定义（应如何调用这些函数/服务），还可以引用触发基于事件的服务调用的事件，以及要等待的事件，这些事件表示这种基于事件的服务调用完成。

在EDA解决方案中，我们通常使用AsyncAPI定义事件驱动的微服务。Serverless工作流“函数”定义支持使用AsyncAPI定义调用语义。有关详细信息，请参见Using Funtions for AsyncAPI Service。

AsyncAPI​

AsyncAPI是一项开源计划，旨在改善事件驱动体系结构（EDA）的当前状态。我们的长期目标是让使用EDA和使用REST API一样容易。包括从文档到代码生成、发现到事件管理。现在应用于REST API的大多数流程也适用于事件驱动/异步API。

详见AsyncAPI官网

工作流示例​

在本示例中，我们构建了上面订单管理系统的事件驱动工作流。

首先，我们需要为我们的微服务应用定义AsyncAPI。

  • 在线商店应用程序
asyncapi: 2.2.0
info:
  title: Online Store application
  version: '0.1.0'
channels:
  store/order:
    subscribe:
      operationId: newStoreOrder
      message:
        $ref : '#/components/NewOrder'

  • 订单服务
asyncapi: 2.2.0
info:
  title: Order Service
  version: '0.1.0'
channels:
  order/inbound:
    publish:
      operationId: sendOrder
      message:
        $ref : '#/components/Order'
  order/outbound:
    subscribe:
      operationId: processedOrder
      message:
        $ref : '#/components/Order'
  • 支付服务
asyncapi: 2.2.0
info:
  title: Payment Service
  version: '0.1.0'
channels:
  payment/inbound:
    publish:
      operationId: sendPayment
      message:
        $ref : '#/components/OrderPayment'
  payment/outbound:
    subscribe:
      operationId: paymentReceipt
      message:
        $ref : '#/components/OrderPayment'
  • 物流服务
asyncapi: 2.2.0
info:
  title: Shipment Service
  version: '0.1.0'
channels:
  shipment/inbound:
    publish:
      operationId: sendShipment
      message:
        $ref : '#/components/OrderShipment'

接下来，定义描述订单管理业务逻辑的订单工作流。

id: storeorderworkflow
version: '1.0'
specVersion: '0.8'
name: Store Order Management Workflow
states:
  - name: Receive New Order Event
    type: event
    onEvents:
      - eventRefs:
          - NewOrderEvent
        actions:
          - eventRef:
              triggerEventRef: OrderServiceSendEvent
              resultEventRef: OrderServiceResultEvent
          - eventRef:
              triggerEventRef: PaymentServiceSendEvent
              resultEventRef: PaymentServiceResultEvent
    transition: Check Payment Status
  - name: Check Payment Status
    type: switch
    dataConditions:
      - name: Payment Successfull
        condition: "${ .payment.status == 'success' }"
        transition: Send Order Shipment
      - name: Payment Denied
        condition: "${ .payment.status == 'denied' }"
        end: true
    defaultCondition:
      end: true
  - name: Send Order Shipment
    type: operation
    actions:
      - eventRef:
          triggerEventRef: ShipmentServiceSendEvent
    end: true
events:
  - name: NewOrderEvent
    source: file://onlineStoreApp.yaml#newStoreOrder
    type: asyncapi
    kind: consumed
  - name: OrderServiceSendEvent
    source: file://orderService.yaml#sendOrder
    type: asyncapi
    kind: produced
  - name: OrderServiceResultEvent
    source: file://orderService.yaml#processedOrder
    type: asyncapi
    kind: consumed
  - name: PaymentServiceSendEvent
    source: file://paymentService.yaml#sendPayment
    type: asyncapi
    kind: produced
  - name: PaymentServiceResultEvent
    source: file://paymentService.yaml#paymentReceipt
    type: asyncapi
    kind: consumed
  - name: ShipmentServiceSendEvent
    source: file://shipmentService.yaml#sendShipment
    type: asyncapi
    kind: produced

对应的工作流图如下：

Workflow Diagram

EventMesh工作流引擎​

在下面的体系结构图中, EventMesh目录, EventMesh工作流引擎 和 EventMesh Runtime在三个不同的处理器中运行。

Workflow Architecture

运行工作流的步骤如下：

  1. 在环境中部署发布者和订阅者应用程序。 使用AsyncAPI描述应用程序API，生成asyncAPI yaml。 使用AsyncAPI在EventMesh目录中注册发布者和订阅者应用程序。

  2. 在EventMesh工作流引擎中注册Serverless工作流DSL。

  3. 工作流引擎从EventMesh目录查询发布服务器和订阅服务器的需要的工作流DSL函数。

  4. 事件驱动App将事件发布到EventMesh Runtime触发工作流。EventMesh工作流引擎发布和订阅事件、编排事件。

EventMesh Catalog 设计​

EventMesh目录存储发布者、订阅者和通道元数据。由以下模块组成：

  • AsyncAPI解析器

    使用AsyncAPI社区提供的SDK (tool list), 解析并验证AsyncAPI yaml输入，并生成AsyncAPI定义。

  • 发布者, 通道, 订阅者模块

    从AsyncAPI定义存储发布者、订阅者和通道信息。

EventMesh工作流引擎设计​

工作流引擎由以下模块组成：

  • 工作流解析器

    使用Serverless Workflow社区提供的SDK(SDKs), 解析和验证工作流DSL输入，并生成工作流定义。

  • 工作流模块

    管理工作流实例的生命周期，从创建、启动、停止到销毁。

  • 状态模块

    管理工作流状态生命周期。支持与事件相关的状态，and the supported state list below is Work-in-Progress.

    工作流状态描述
    Operation执行Actions中定义的AsyncAPI函数
    Event检查定义的事件是否匹配，如果匹配，执行定义的AsyncAPI函数
    Switch检查事件是否与事件条件匹配，并执行定义的AsyncAPI函数
    Parallel并行执行定义的AsyncAPI函数
    ForEach迭代输入集合并执行定义的AsyncAPI函数

  • 行为模块

    管理函数中的行为。

  • 函数模块

    通过在EventMesh Runtime中创建发布者和/或订阅者来管理AsyncAPI函数，并管理发布者/订阅者生命周期。

    AsyncAPI 操作EventMesh Runtime
    PublishPublisher
    SubscribeSubscriber

  • 事件模块

    使用工作流DSL中定义的规则管理CloudEvent数据模型，包括事件过滤器、关联和转换。

  • 重试模块

    管理事件发布到EventMesh Runtime的重试逻辑。