如何构建以事件驱动型实时信息系统

图：实时信息系统解决方案架构

如上图所示，各类事件通常来源于包括移动和Web应用在内的不同渠道。其中：

• 接收事件组件，负责在任何给定的时间内处理大量的数据，其速率从每秒上千个事件、到每秒数上百万个事件不等。在接收到事件之后，该组件会将其传递给对应的处理组件。
• 事件处理组件，负责对事件中带有的数据进行操作，包括：过滤，清理，转换和汇总等。根据组件的行为，事件处理既可以独立于事件接收组件、被单独地执行，也可以作为相关的操作来完成。事件处理组件可以将原始事件存储在各种数据商店(data store)中，以便进行审核(如果未被接收组件处理的话)，并且可以根据用例，将计算结果存储在那些单独的数据商店之中。因此，大多数事件都能够被实时处理，并发布到事件发布组件中，以传递给使用者(consumer)。
• 事件发布组件，负责将处理后的数据实时地推送给使用者。这些使用者既可以是移动或Web应用，又可以是对已处理事件起作用的其他系统。除了这些实时事件的发布之外，在某些用例中，我们还需要通过HTTP通道，以同步、或请求-响应的方式，将处理后的摘要信息发布给移动和Web应用。

上述参考架构可用于那些需要处置现场正在发生的事件，并将其发布到后台应用等多种用例中。例如：在出现紧急情况时，现场人员可以实时地发送有关需求与状况的细节信息，而后台团队则能够毫不拖延地进行必要的物资调配与派送。此外，我们也可以使用此类架构来构建农业的供应链。例如：农民们通过运货车辆将农作物运送到连锁超市。农民可以在农作物准备就绪时，更新其详细信息。而超市后台团队则会实时地从各个位置获取更新，并安排车辆及时地收集农作物，以避免延迟。

使用WSO2和Kafka的参考架构

消息代理是将消息发送者与接收者相分离的组件。目前，市场上有很多消息代理类产品，它们各有优、缺点。其中最流行的当属Kafka、NATS和RabbitMQ。当然，Kafka也可以作为NATS和RabbitMQ的最佳功能性代理。

在此，我们选择Kafka作为事件消息的代理;选择功能丰富、简单且开源的WSO2Streaming Integrator作为事件处理器;使用既支持流媒体、又支持REST风格的WSO2 API Manager，作为事件发布者。当然，这些组件也可以被市场上的其他类似工具所替换。下图展示了构成实时事件驱动型信息系统的各个组件，及其相互连接。

图：具有代理和WSO2平台的实时事件驱动型信息系统

在该架构中，事件代理会接收来自移动和Web应用等源头的事件负载。WSO2 Streaming Integrator会处理这些事件，然后将各种结果事件通过WebSocket连接，发布到WSO2 API Manager上。WSO2 API Manager拥有一个公布给网关的WebSocket API，诸如移动和Web之类的consumer(消费者)应用会使用该API，实时地接收各种事件。同时，WSO2 Streaming Integrator可以将原始事件和汇总的结果，通过标准的REST API，从WSO2 API Manager处公布给相关的consumer。下图对上述架构进行了细化。

图：具有Kafka和WSO2平台详细信息的实时事件驱动型信息系统

如上图所示，事件源通过Kafka客户端，将事件发布到Kafka代理中那些可用的topic(主题)处。WSO2 Streaming Integrator不但可以订阅这些topic，还能通过已配置的Kafka源，实时地使用来自Kafka的各种事件。由Siddhi语言编写的各项操作将处理这些事件，并传递给诸如WebSocket之类的事件sink(接收器)。同时，WSO2 SI会按需通过各种数据商店(data stores)，将事件存储到对应的数据库表中。

WSO2 API Manager通过WebSocket API来将WebSocket sink的详细信息配置到API的端点上。据此，那些使用WebSocket API的客户端应用将能实时地接收到已经处理的各种事件。

同时，那些已处理的信息和原始事件会被存储到一个通过WSO2 Enterprise Integrator公布了REST数据服务的数据库中。此处的数据服务是通过将WSO2 API Manager作为受保护的REST API予以公布，并通过客户端应用实现同步通信的服务。此外，作为一种能够支持大多数企业系统需求的成熟架构，我们可以通过扩展，来支持诸如：混合集成需求、API管理平台等多种企业用例。

只有WSO2平台，没有Kafka的参考架构

如果贵组织刚开始着手构建实时的事件驱动型信息系统，而且数据负载量并不大的话，那么就可以仅使用WSO2平台，来构建前文提到的精简版架构。下图展示了一种没有消息代理的实现方式。

图：具有WSO2平台的实时事件驱动型信息系统

该架构与前文提到的架构之间唯一的区别在于：虽然缺少事件代理，但是客户端应用能够通过HTTP的调用，将事件直接发送到WSO2 Streaming Integrator处。当然，由于该架构没有消息代理，因此WSO2 SI需要将原始事件存储在数据库中，以供各项审核。而它的其余功能则与前文的架构相同。下图展示了该架构的详细组成结构。

图：具有WSO2平台详细信息的实时事件驱动型信息系统

如上图所示，WSO2 SI被配置为通过HTTP接口来接收事件。而Siddhi应用中的HTTP源则被配置为通过不同的操作，来处理各种事件，然后发布到WebSocket sink中。同时，各种原始事件通过数据商店被存储在数据库中，并将各种聚合的结果通过不同的数据商店存储到另一个数据表里。除此之外，该系统的其余功能与前文提到的基于代理的实现方式基本一致。

从Kafka到Websocket Siddhi应用的示例代码

下面我们将给出一个Siddhi的应用示例。它能够从Kafka的topic中读取事件，并通过WebSocket服务器，将各种事件发布(或输出)到某个日志sink处。当然，在发布之前，它会对每个事件进行简单地检查(或筛选)，以确保其数量小于500。具体代码请参见--https://gist.github.com/chanakaudaya/efe8dfed2558811f0316a7839dbfef57。其中，您可以找到有关如何使用Streaming Integrator，来设置Kafka的详细示例。同时，您也可以通过文档链接--https://ei.docs.wso2.com/zh_CN/latest/streaming-integrator/examples/working-with-kafka/，来试运行该Siddhi应用。

如何创建连接到WebSocket端点的WebSocket API

如下图所示，您可以通过WSO2 API Manager的发布者(publisher)接口来创建WebSocket API，并使用WS服务器将这些事件发布到客户端。

您可以在下图的上部菜单中选择“设计新的WebSocket API”(或“创建API”)，然后在下一个窗口中提供详细的信息。

接着，您可以选择“创建并发布”选项，将WebSocket API推送到开发人员的门户(portal)处，以便用户在其中使用有效的OAuth2令牌。

通过参考文档链接-- https://apim.docs.wso2.com/zh-CN/latest/learn/tutorials/create-and-publish-websocket-api/#create-and-publish-a-websocket-api，您可以逐步了解到如何创建WebSocket API，并能够试运行其客户端的示例。