使用 Angular/Ngrx 和 Vert.x 构建实时 Web 应用

欢迎来到实时 Web 的世界!现在是从传统的同步 HTTP 请求/响应架构转移到已连接客户端的响应式应用程序的时候了（这只是一句话中的很多流行语）！

图片来源:  https://www.voxxed.com

MeteorJS 可以用来构建这样的应用程序（v1.0发布于2014年10月）： 一个全栈 Javascript 平台，用于构建已连接客户端的应用程序 。它允许 JS 开发人员立即构建和部署 web 和移动端应用程序（iOS/Android），并且可以使用统一的后端+前端。这是一个非常雄心勃勃的方法，但它需要一个非常虔诚和高度耦合的 JS 技术堆栈，它仍然是一个很好的框架。

此外，我们是后端的 Java 商店。在 AgoraPulse 中，我们严重依赖于：

• 应用于 JS 前端的 Angular 和  Ionic （包括基于  Angular 和  Ionic 的共享业务/数据架构），

• 用于JVM后端的 Groovy 和  Grails 。

所以我的问题是...

构建基于 JVM 的后端的实时 Angular 客户端应用程序的最佳选择是什么？

我们的要求是非常简单的。我们不需要完整的 Meteor 端到端应用模型。我们只是希望能够：

1. 基于 JVM 构建一个使用 事件总线的响应式应用程序

2. 将 事件总线扩展 到浏览器，以便能够从 Angular app 上发布/订阅实时事件。

服务器端 (JVM)

Reactive app 是当今的热门话题，并且有许多优秀的库/平台可以用于在 JVM 上构建这种类型的事件驱动框架：

• Vert.x ,

• TypeSafe 平台 (Play, Akka, Scala),

• Spring  Project Reactor  (已集成到 Grails 3.0 中)

注意：对于 Grails， Spring Websocket Grails 插件 看起来很有意思，特别是在你使用了完整的 Spring 生态系统时：Mike Plummer 有一个很好的关于  Grails 3.0  中的 WebSockets  的 例子和博客文章。

客户端 (JS)

ReactJS 和  Angular 是目前用于构建现代化 JS 应用程序两个最受欢迎的框架。大多数平台使用  SockJS 处理实时连接：

• Vertx-web 提供了一个带有事件总线桥的 SockJS 服务器实现和一个  vertx-evenbus.js  客户端库（非常易于使用）

• Spring 通过 Spring Messaging 和 Websocket 库提供了 Websocket SockJS 支持（ 相关示例在此 ）

最终选择： Vert.x + Angular

最后，我选择了使用 Vert.x 来完成试验，因为它具有优秀的 Groovy 支持、分布式事件总线、可扩展性和易用性。

我非常喜欢这种方式。让我向你展示下我的试验结果，这是在 AgoraPulse v6.0 中即将推出的实时功能的起源！

Vert.x 是什么?

Vert.x 是用于在 JVM 上构建多应用程序反应性应用程序的 工具 包。

Vert.x 应用程序由一个更松耦合的组件组成，称为 Verticles，并行运行。

Verticles 是由 Vert.x 部署和运行的代码块。 Verticle 可以用 Vert.x 支持的任何语言编写，单个应用程序可以包含以多种语言。

图片源： https://www.slideshare.net/

应用程序通常由同一个 Vert.x 实例中运行的许多 verticle 实例组成。 不同的 verticle 实例通过在事件总线上发送消息来相互通信。

为什么选择 Vert.x？

和其它响应式平台类似，Vert.x 是 事件驱动 且 非阻塞 。它的伸缩性很好(甚至比 Node.js 还好)。

与其它响应式平台的不同之处在于，Vert.x 支持多语言：你可以在多种语言中使用 Vert.x，包括 Java、JavaScript、Groovy、 Ruby 、Ceylon、Scala 和 Kitlin。

与 Node.js 的不同之处在于，Vert.x 以 通用 工具为目的，并非固定于某种用途。它是一个多用途的平台，可以做很多事情：包括简单的网络工具，复杂的现代 Web 应用，HTTP/REST 微服务，或者丰富的后端消息总线应用。

它像其它响应式平台一样，一开始阅读文档的时候都会感到害怕...;) 但是你一旦开始使用它，它总是保持 有趣 而且 易用 ，特别在 Groovy 中使用！Vert.x 能实实在在地让你构建系统，而不需要去纠结各种复杂的问题。

以我为例，我主要对分布式事件总线(Vert.x 的一个核心特性)感兴趣。

我们建立一个原型来验证我们的方法，这个原型的目标如下：

• 在多个客户端连接中共享和同步一个普通的(基于 Ngrx 的)状态，以及

• 在之个客户端连接中实时分发(基于 Ngrx)行为动作，它们会影响本地状态/归约器。

注意： @ngrx/store  是由 RxJS 推动的 Angular 应用的状态管理，它的灵感来自 Redux。这是目前 Angular 应用中最流行的组织复杂业务逻辑的方法。

来源： https://www.smashingmagazine.com

概念性验证

这里是我们最初概念性验证的储存仓库：

http://github.com/benorama/ngrx-realtime-app

这个仓库包括两个独立项目：

• Vert.x server app, 基于Vert.x (版本 3.3) 由Gradle管理, 主要使用Groovy语言垂直开发。

• Angular client app, 基于Angular （版本 4.0.1），由带状态的Angular CLI管理，以及基于@ngrx/store

我们提供 counter 示例代码 (actions 和 reducers) 供您试用，来源@ngrx/store。

Counter 客户端业务逻辑基于：

• CounterState 接口，counter 状态模型

• counterReducer reducer，基于被分发的 actions 的counter状态管理

• 增， 减 和 重置 计数器行为。

状态由服务器端的一个简单独立类 CounterService 维护。

class CounterService {
    static INCREMENT = '[Counter] Increment'
    static DECREMENT = '[Counter] Decrement'
    static RESET = '[Counter] Reset'
    int total = 0
    void handleEvent(event) {
        switch(event.type) {
            case INCREMENT:
                total++
                break
            case DECREMENT:
                total--
                break
            case RESET:
                total = 0
                break
        }
    }
}

通过请求(Request)/响应(Response)初始化客户端状态

来源:  https://www.slideshare.net/

初始状态由简单的 请求/响应 (或者发送/回复)在事件总线上初始化。

客户端连接上之后，会发送一个请求到位于 counter::total 的事件总线。服务器直接使用  CounterService 的 total 值进行回复，客户端使用这个回复的 total 值在本地触发一个 reset 行为。

下面是抽取出来的相关代码(来自 AppEventBusService )：

initializeCounter() {
    this.eventBusService.send('counter::total', body, (error, message) => {
    // Handle reply
    if (message && message.body) {
            let localAction = new CounterActions.ResetAction();
            localAction.payload = message.body; // Total value
            this.store.dispatch(localAction);
        }
    });
}

通过 Publish/Subscribe（发布/订阅）进行 Action 分发

源:  https://www.slideshare.net

使用 publish/subscribe模式 处理Action分发/同步.

counter action 从客户端发送到事件总线的 counter::actions  地址上。

任何已订阅  counter::actions  地址的客户端将收到这些操作，并在本地重新处理它们以改变应用程序的状态 /reducer。

以下是相应代码的摘录（摘自 AppEventBusService）：

publishAction(action: RemoteAction) {
    if (action.publishedByUser) {
        console.error("This action has already been published");
        return;
    }
    action.publishedByUser = this.currentUser;
    this.eventBusService.publish(action.eventBusAddress, action);
}
subscribeToActions(eventBusAddress: string) {
    this.eventBusService.registerHandler(eventBusAddress, (error, message) => {
        // Handle message from subscription
        if (message.body.publishedByUser === this.currentUser) {
            // Ignore action sent by current manager
            return;
        }
        let localAction = message.body;
        this.store.dispatch(localAction);
    });
}

时间总线发布逻辑是通过一个简单的 NgrxEffect 实现的。任何扩展自 RemoteAction 类的类都将被发布到事件总线上。

@Injectable()
export class AppEventBusEffects {
  
    constructor(private actions$: Actions, private appEventBusService: AppEventBusService) {}
    // Listen to all actions and publish remote actions to account event bus
    @Effect({dispatch: false}) remoteAction$ = this.actions$
        .filter(action => action instanceof RemoteAction && action.publishedByUser == undefined)
        .do((action: RemoteAction) => {
            this.appEventBusService.publishAction(action);
        });
  @Effect({dispatch: false}) login$ = this.actions$
        .ofType(UserActionTypes.LOGIN)
        .do(() => {
            this.appEventBusService.connect();
        });
}

你可以通过在两个单独的浏览器窗口中启动本地服务器和客户端应用程序来查看所有这些结果。

福利：演示的 app 还包括基于事件总线连接状态的用户状态展示（离线/在线）。

计数器状态会在已连接的客户端之间共享和同步，并且每个本地动作都实时分发给其他客户端。

任务完成！

注意：这个 app 使用了我们自己的 Typescript 版本，这是基于官方 JS Vertx EventBus 客户端的，可以在 这里 找到：欢迎任何的反馈和改进建议！