如何使用 Google App Engine Data Store (HRE) 创建可靠的聊天？

Question

在我的应用程序中，我想要一个实时聊天功能 - 多人（可能 5 人或更多）可以同时聊天。

我正在使用基于 Java 的 Google App Engine - 这真的是我第一次尝试使用 GAE Datastore，我已经习惯使用 Oracle/MySQL，所以我认为我的策略是错误的。

注意：为简单起见，我省略了任何验证/安全检查 在一些名为 WriteMessage 的 servlet 中，我有以下代码

Entity entity = new Entity("ChatMessage");
entity.setProperty("userName", request.getParameter("userName"));
entity.setProperty("message", request.getParameter("message"));
entity.setProperty("time", new Date());
DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
datastore.put(entity);

在其他一些名为 ReadMessages 的 servlet 中，我有以下代码

String id = request.getParameter("id");
DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
Query query = new Query("ChatMessage");
if (id != null) {
  // Client requested only messages with id greater than this id
  Filter idFilter = new FilterPredicate(Entity.KEY_RESERVED_PROPERTY,
        FilterOperator.GREATER_THAN,
        KeyFactory.createKey("ChatMessage", Long.parseLong(id)));
  query.setFilter(idFilter);
}
PreparedQuery pq = datastore.prepare(query);
JsonArray messages = new JsonArray();
for (Entity result : pq.asIterable()) {
  JsonObject jmsg = new JsonObject();

  // Client will use this id on the next request to read to poll only
  // "new" messages
  jmsg.addProperty("id", result.getKey().getId());
  jmsg.addProperty("userName", (String) result.getProperty("userName"));
  jmsg.addProperty("message", (String) result.getProperty("message"));
  jmsg.addProperty("time", ((Date) result.getProperty("time")).getTime());
  messages.add(jmsg);
}
PrintWriter out = response.getWriter();
out.print(messages.toString());

在 javascript 客户端代码中 - 每当用户提交新消息时都会调用 WriteMessage servlet - 并且每秒调用一次 ReadMessages servlet 以获取新消息。

为了优化，javascript 将在后续请求中将它收到的最后一条消息的 id（或者可能是迄今为止它收到的最高 id）发送到 ReadMessage，以便响应仅包含它的消息没见过。

起初这一切似乎都有效，但我认为这段代码可能存在一些问题。

以下是我认为的错误：

• 可能无法读取某些消息，因为我依赖 ChatMessage 的密钥的 id 来过滤 JS 客户端之前已经看到的消息 - 我认为这不可靠吗？

• 某些写入可能会失败，因为可能同时有 5 或 6 个传入写入 - 我的理解是，如果每秒写入太多，这可能会导致 ConcurrentModificationException。

• 传递给实体的日期是应用程序服务器上 JRE 的当前日期 - 也许我应该在 SQL 中使用类似“sysdate()”的东西？我不知道这是否真的是一个问题。

我怎样才能修复代码以便：

1. 所有聊天消息都将被写入 - 是否最好进行故障转移，以便如果请求失败，javascript 将重新尝试直到成功？

2. 将读取所有聊天消息（无例外）

3. 清理旧消息，只存储 1000 条左右的消息

Answer 1

当有人在向 SO 发布问题之前实际解决问题时，这有点令人耳目一新。

虽然您确实列出了您在使用方法时遇到的一系列有效问题，但我建议您最大的问题是成本。您正在为每条聊天消息添加一个新实体，此外，该实体需要被索引。因此，您正在谈论针对发送的每条消息的多个写入操作。您还必须为您删除的每个实体付费，因此您必须付费才能清理。

在您的设计的有利方面，您没有使用事务或祖先来创建您的实体，因此您不应该达到写入性能限制。

在读取方面，您每条消息读取一个实体，因此成本也会在那里累加。您在没有事务或祖先查询的情况下进行查询这一事实意味着您在查询时可能看不到最新的 ChatMessage 实体。

此外，与 SQL 不同的是，GAE 数据存储区 id 不是单调递增的，因此通过 id GREATER_THAN 进行查询是行不通的。

现在提出建议。我警告你，这将是很多工作。

1. 尽量减少您使用的实体数量。不要为每个消息添加一个新实体，而是使用一个更大的实体，每个实体存储多个消息。

2. 不是查询消息实体，而是通过键获取它们。按键获取实体将为您提供高度一致的结果，而不是最终一致的结果。如果您想确保阅读所有最新的聊天消息（无例外），这一点很重要

这确实带来了两个您需要处理的新问题：

• 如果对同一个实体进行多次写入，您将达到某种写入性能限制。

• 由于您的实体可能会变得很大，因此您需要处理此案例以确保它们不会超出 1MB 的限制。

您需要两种实体种类。您将需要一个存储多条消息的 MessageLog 类型。您可能希望将消息存储为 MessageLog 中的列表。对于给定的聊天，您将需要多个 MessageLog 实体，主要是为了提高写入性能。 （搜索“Google App Engine Sharding”了解更多信息）。

您需要一个基本上存储 MessageLog 键列表的 Chat Kind。这允许进行多个聊天。您最初的实现似乎只有一个全局聊天。或者，如果您愿意，只需使用 Chat 的单个实例。

这些都不需要索引，因为您将通过 Key 获取所有内容。这将降低成本。

当您开始新的聊天时，您将根据您期望的性能创建许多 MessageLog 实体。您期望的每秒写入 1 个实体。如果聊天中有更多人，我会创建更多消息日志。然后创建一个 Chat 实体并将 MessageLog 键列表存储在其中。

在写消息时，您将执行以下操作： - 按键获取适当的聊天实体，您现在有一个消息日志列表 - 选择一个 MessageLog 来分配负载，这样所有的写入都不会命中同一个实体。选择一种可能有多种技术，但在本例中，随机选择一种。 - 格式化新消息并将其插入 MessageLog。此时，您还可以考虑将旧消息放入 MessageLog 中。您还希望进行一些安全检查以确保 MessageLog 在 1MB 实体大小限制内。 - 编写消息日志。这应该只产生 1 次写入操作，而不是用于写入新实体的最少 3 次写入操作。 建议：将消息附加到包含整个聊天日志的给定聊天的 memcache 条目。

在阅读时，您将执行以下操作： 推荐：首先检查给定聊天的 memcache 条目，如果存在，则返回，完成。 - 按键获取适当的聊天实体，您现在有一个消息日志列表 - 按键获取所有消息日志。现在，您在聊天中拥有所有消息，并且它们是最新的。 - 解析所有消息日志，并重建整个聊天日志。 推荐：将重建的消息日志存储在 memcache 中，这样您就不必再这样做了。 - 返回重建的聊天记录。

考虑使用 Channel API 将消息发送给观众。通过这种方式，查看者可以比每秒一次更快地接收消息。我个人发现 Channel API 不是 100% 可靠的，所以我不会完全摆脱轮询，但你可能会接受每 30 秒轮询一次作为备份。

想象一下其中包含 100 条消息的聊天。您最初的计划将花费大约 101 次阅读操作来阅读 100 条消息。在这种新方法中，您将拥有 5-10 个 MessageLog 实体，因此成本将是 6-11 个读取操作。如果你得到一个内存缓存命中，你不需要任何读取操作。但是您必须编写代码来从多个 MessageLog 对象重建聊天日志。