数据存储查询超时

Question

我在 java8 运行时环境中使用来自自动缩放应用引擎实例的 objectify v5.1.11。

我有一个 API，IOT 设备会定期调用它来上传统计信息。在这个 API 中，我将一个实体插入到数据存储中以存储统计信息。该实体使用自动生成的数据存储 ID。实体定义如下：

@Entity(name = "Stats")
public class StatsEntity {
    @Id
    private Long statisticsId;

    @Index
    private Long deviceId;

    @Index
    private String statsKey;

    @Index
    private Date creationTime;
}

但我需要在插入实体之前检查重复项。我切换到自定义生成的（字符串）ID。我想出了一种机制，将设备提供的deviceId 附加到statsKey（设备内的每个统计信息唯一）字符串以生成 ID。

这是为了避免eventual consistency behaviour 如果我使用查询来检查实体是否已经存在。由于通过 ID 获取是高度一致的，因此我可以使用它来检查重复项。

还有另一个 API 可以获取设备上传的统计信息。在此 API 中，我通过过滤 deviceId 并按 creationTime 以降序（最新的优先）排序来列出实体，页面大小为 100。此请求超时，因为请求超过了 appengine 的 60 秒限制。我在日志中看到以下异常：

Task was cancelled.
java.util.concurrent.CancellationException: Task was cancelled.
    at com.google.common.util.concurrent.AbstractFuture.cancellationExceptionWithCause(AbstractFuture.java:1355)
    at com.google.common.util.concurrent.AbstractFuture.getDoneValue(AbstractFuture.java:555)
    at com.google.common.util.concurrent.AbstractFuture.get(AbstractFuture.java:436)
    at com.google.common.util.concurrent.AbstractFuture$TrustedFuture.get(AbstractFuture.java:99)
    at com.google.appengine.tools.development.TimedFuture.get(TimedFuture.java:42)
    at com.google.common.util.concurrent.ForwardingFuture.get(ForwardingFuture.java:62)
    at com.google.appengine.api.utils.FutureWrapper.get(FutureWrapper.java:93)
    at com.google.appengine.api.datastore.FutureHelper.getInternal(FutureHelper.java:69)
    at com.google.appengine.api.datastore.FutureHelper.quietGet(FutureHelper.java:33)
    at com.google.appengine.api.datastore.BaseQueryResultsSource.loadMoreEntities(BaseQueryResultsSource.java:243)
    at com.google.appengine.api.datastore.BaseQueryResultsSource.loadMoreEntities(BaseQueryResultsSource.java:180)
    at com.google.appengine.api.datastore.QueryResultIteratorImpl.ensureLoaded(QueryResultIteratorImpl.java:173)
    at com.google.appengine.api.datastore.QueryResultIteratorImpl.hasNext(QueryResultIteratorImpl.java:70)
    at com.googlecode.objectify.impl.KeysOnlyIterator.hasNext(KeysOnlyIterator.java:29)
    at com.google.common.collect.Iterators$5.hasNext(Iterators.java:580)
    at com.google.common.collect.TransformedIterator.hasNext(TransformedIterator.java:42)
    at com.googlecode.objectify.impl.ChunkIterator.hasNext(ChunkIterator.java:39)
    at com.google.common.collect.MultitransformedIterator.hasNext(MultitransformedIterator.java:50)
    at com.google.common.collect.MultitransformedIterator.hasNext(MultitransformedIterator.java:50)
    at com.google.common.collect.Iterators$PeekingImpl.hasNext(Iterators.java:1105)
    at com.googlecode.objectify.impl.ChunkingIterator.hasNext(ChunkingIterator.java:51)
    at com.ittiam.cvml.dao.repository.PerformanceStatsRepositoryImpl.list(PerformanceStatsRepositoryImpl.java:154)
    at com.ittiam.cvml.service.PerformanceStatsServiceImpl.listPerformanceStats(PerformanceStatsServiceImpl.java:227)

设备提供的statsKey 是基于时间的，因此单调增加（步长增加 15 分钟），这对于link 来说是不好的。 但是我的流量不足以保证这种行为。每台设备每 15 分钟发出 2 到 3 个请求，大约有 300 台设备。 当我尝试列出自切换到自定义 ID 以来未发出任何请求的设备的实体时，我仍然会发现此问题。

编辑

我列出实体的代码如下：

Query<StatsEntity> query = ofy().load().type(StatsEntity.class);

List<StatsEntity> entityList =
        new ArrayList<StatsEntity>();

query = query.filter("deviceId", deviceId);

query = query.order("-creationTime");

query = query.limit(100);

QueryResultIterator<StatsEntity> iterator = query.iterator();

while (iterator.hasNext()) {
    entityList.add(iterator.next());
}

Answer 1

这个错误通常是因为write contention. 而发生的 有多种方法可以解决这个问题：

• 一个查询只存在 30 秒，但您可以通过将 API 转换为任务队列来扩展它。通常处理此类写争用问题时，您应该始终使用持续约 10 分钟的任务队列。
• 如果可能，请缩小您的实体组。

您可以找到更多方法here.

希望这能回答你的问题！！！