数据网织和基于网格的分布式计算

随着软件系统越来越复杂,用户越来越多，需求也更加地“大数据”化，传统的数据库可能会成为系统的性能瓶颈。面临这种现实的企业别无其他选择，只能在计算能力上寻求解决。数据网织(Data Fabric)--有时也称为网格计算¹²--可以满足这种性能上的需求。

¹². 这并不是说网络和网格是相同概念, 但是对于那些从整体角度来看这种框架的人来说，两者指的是同样东西，对于市场和营销人来说，他们也倾向于这两者同等对待. 无论如何, 本节将使用"数据网织"，因为通常它所表示的功能比网络计算要多。 ↩

数据网织的一个好处是它对领域模型提供了自然的支持，几乎消除了所有的阻抗失配。事实上，分布式缓存可以非常容易地对领域模型进行持久化，此时可以将它看成是一种聚合存储(Aggregate Store)¹³。简单地说，在数据网织中，聚合即是基于图的缓存中的值部分¹⁴,而聚合的唯一标识则是标识键。这里的键即是聚合的唯一标识。聚合的状态将被持久化为二进制数据或文本数据，这些数据便是值部分的内容：

¹³. Martin Fowler最近正在提出使用"聚合存储", 虽然这个概念已经存在了一段时间。 ↩

¹⁴. 在Gemfire中，这称为区域(region),而在Coherence中，则称为缓存(cache)，在本书中，我将使用后者。 ↩

String key = product.productld().id();

byte[] value = Serializer.serialize(product);

// region (GemFire) or cache (Coherence)
region.put(key, value);

因此，数据网织能够很好地在技术层面上与领域模型保持一致，从而在很大程度上缩短开发周期¹⁵。

¹⁵. 有些NoSQL存储也可以作为自然的聚合存储,它们也可以用来简化DDD的技术实现。 ↩

上例向我们展示了数据网织是如何将领域模型存放在缓存中的。那么，有了这样的数据网织，我们将如何使用长时处理过程来支持CQRS架构和事件驱动架构呢？

数据复制

考虑一个内存数据缓存，我们可以立刻想到由缓存失败造成系统状态丢失的种种可能性。这是一个真实存在的问题，但是在允许数据复制的数据网织中，这并不是什么大问题。

在使用“一个缓存对应一个聚合”的策略时，让我考虑一下由数据网织提供的内存缓存。在这种情况下，某个聚合类型的资源库由一个专门的缓存提供支持。只支持单个节点的缓存是很容易失效的。然而，一个具有多节点缓存的数据网织则是可靠的。根据节点有可能的失效数目，你可以选择不同层次的数据冗余性。由于多个缓存节点的存在，失效的几率将随之变小。同时你可以从数据冗余性中获得更高的性能，因为性能受到节点数目的影响。

对于缓存冗余性的工作机制，这里有一个例子：其中一个节点作为主缓存，其他节点作为二级缓存。如果主缓存失效，其中一个二级缓存将会成为新的主缓存。当先前失效的主缓存恢复之后，新主缓存中的数据将被复制到恢复后的缓存中，此时该恢复后的缓存将变为二级缓存。

这种做法的另一个好处是，它可以保证对数据网织所发出事件的正确投递。因此，在聚合更新之后，数据网织所发出的事件是不会丢失的。显然，对于保存核心的领域模型来说，缓存冗余性和数据复制扮演着重要的角色。

事件驱动网织和领域事件

数据网织可以很好地支持事件驱动架构风格，因为它能确保对事件的投递。大多数数据网织都有内建的事件支持，即可以对缓存层面和入口层面上所发生的操作自动地发出事件通知。这些事件不应该和领域事件产生混淆。比如，一个缓存层面的事件用于通知诸如“重新初始化缓存”这样的操作;一个人口层面的事件描述诸如“创建人口和更新入口”等操作。

数据网织是支持开放架构的，因此应该有种方法可以从聚合中直接发布领域事件。此时，领域事件可能需要继承框架中的某种事件类型，比如GemFire中的EntryEvent，但是相比起这些领域事件所提供的强大功能来说，这种继承只是很小的代价。

那么，我们究竟应该如何在数据网织中使用领域事件呢？就像在领域事件(8 )中所讨论到的一样，此时的聚合应该使用一个简单的DomainEventPublisher组件。对于数据网织的缓存来说，这个发布组件可能只是简单地将事件放在某个特定的缓存中。此后，缓存事件(Cached Event)将通过同步或异步的方式送达订阅方。因此，为了不浪费内存，我们可以在得到所有订阅方的应答之后，将缓存事件从缓存图中移除掉。当然，当事件被发送到消息队列或用于刷新CQRS的查询模型时，这个事件只会得到一次应答。

对于领域事件的订阅方来说，他们可以将事件用于同步地更新其他相关的聚合，由此最终一致性也得到了保证。

持续查询

有些数据网织支持一种名为持续查询(Continuous Query)的事件通知。客户端可以向数据网织注册一个查询，当对缓存的修改可能影响到查询结果时，客户端将自动接收到事件通知。持续查询可以用于用户界面组件，此时用户界面组件可以监听那些有可能影响用户视图的修改。

CQRS可以很好地与持续查询相结合，此时我们假设查询模型由数据网织维护。此时，我们不用等到视图表更新，而可以直接通过持续查询来及时地更新视图。在下面的例子中，一个客户端注册了一个GemFire持续查询事件：

CqAttributesFactory factory = new CqAttributesFactory();

CqListener listener = new BacklogItemWatchListener();

factory.addCqListener(listener);

String continuousQueryName = "BacklogItemWatcher";

String query = "select * from /queryModelBacklogItem qmbli "
        + "where qmbli.status = 'Committed'";

CqQuery backlogItemWatcher = queryService.newCq(
        continuousQueryName, query, factory.create());

现在，数据网织便可以通过客户端中的一个回调对象来更新CQRS查询模型，该回调对象由CqListner创建。同时，所传给回调对象的数据还包含了新增的或更新的元数据。

分布式处理

数据网织的另一个功能是，它可以在所有复制缓存范围内完成分布式处理，然后将处理结果聚合到一起发给客户端。这使得数据网织可以用于事件驱动的、分布式的并行处理过程中，比如长时处理过程。

为了演示以上功能，我们来看看GemFire和Coherence的一些实现细节。此时，长时处理过程的执行器可以实现为GemFire中的一个函数(Function),或者Coherence中人口处理器(Entry Processor)。这两者都实现了命令模式(Command)[Gamma et al.]，用于在分布式的复制缓存中执行命令。(你也可以将此想为领域服务，但是它并不会以领域为中心）为了使概念一致，我们将此功能统一称为函数。一个函数可以选择性地使用一个过滤器来去除掉那些不满足条件的聚合实例。

让我们来看看一个实现了长时处理过程的函数是如何处理先前的电话号码例子的。这个处理过程将在复制缓存中并行执行，使用的是GemFire函数：

public class PhoneNumberCountSaga extends FunctionAdapter {
    @Override
    public void execute(FunctionContext context) {
        Cache cache = CacheFactory.getAnyInstance();
        QueryService queryService = cache.getQueryService();

        String phoneNumberFilterQuery = (String) context.getArguments();
        ...
        // 伪代码
        // -执行函数以获取MatchedPhoneNumbersCounted.
        // -通过调用aggregator.sendResult (MatchedPhoneNumbersCounted)将答案发送给聚合器
        // -执行函数以获取 AllPhoneNumbersCounted.
        // -通过调用aggregator.sendResult (AllPhoneNumbersCounted)将答案发送给聚合器.
        // -聚合器从每个分布式的函数返回中自动地收集答案，然后将聚合
        // 之后的单一答案发送给客户端
    }
}

客户端可以通过以下方式来并行地执行一个长时处理过程:

PhoneNumberCountProcess phoneNumberCountProcess =
         new PhoneNumberCountProcess();

String phoneNumberFilterQuery =
        "select phoneNumber from /phoneNumberRegion pnr "
        + "where pnr.areaCode = '303'";

Execution execution =
        FunctionService.onRegion(phoneNumberRegion)
                .withFilter(0)
                .withArgs(phoneNumberFilterQuery)
                .withCollector(new PhoneNumberCountResultCollector());

PhoneNumberCountResultCollector resultCollector =
         execution.execute(phoneNumberCountProcess);

List allPhoneNumberCountResults = (List) resultsCollector.getResult();

当然，我们可以将以上处理过程变得更加复杂，也可以将其简化。这也说明，一个长时处理过程不见得一定是事件驱动的，而是可以通过其他并行处理方式来完成的。关于基于数据网织的分布式和并行处理的更多内容，请参考[GemFireFunctions]