如何在 Google App Engine / Google Data Store (JDO) 中实现消息队列

Question

我想为部署在 Google App Engine 中的应用程序实现消息队列 (F.I.F.O)。该应用程序使用 JDO 访问 Google DataStore。我公开了一个可以从任何移动设备调用的 REST API。下图应该清楚地表明我想要实现的目标。

移动设备池调用 REST API 将消息排入队列。消息实体包含创建时间戳和消息状态字段。消息状态最初设置为活动。另一个移动设备池将调用 REST API 来获取消息。在那一刻，查询数据存储以获取按创建时间戳升序排序的 ACTIVE 消息实体。然后将消息状态设置为 DELIVERED 并将消息返回到设备。

当来自消费者设备的并​​发请求时，就会出现问题。多个设备收到相同的消息。

我有一个模型类消息

@PrimaryKey
@Persistent(valueStrategy = IdGeneratorStrategy.IDENTITY)
@Extension(vendorName = "datanucleus", key = "gae.encoded-pk", value = "true")
private String id;
@Persistent
private String message;
@Persistent
private Date created;
@Persistent
private MessageStatus status;

MessageStatus 枚举具有以下状态

    public enum MessageStatus {
    ACTIVE, DELIVERED, ACKNOWLEDGED
}

我有一个数据访问类 MessageDAO

public Message fetchMessage() {
        PersistenceManager pm = PMF.get().getPersistenceManager();
        Message message = null;
        Transaction transaction = null;
        try {
            transaction = pm.currentTransaction();
            transaction.setIsolationLevel(Constants.TX_READ_COMMITTED);
            transaction.begin();            

            Query query = pm.newQuery(Message.class);           
            query.setFilter("status == 'ACTIVE'");          
            query.setOrdering("created asc");


            List<Message> messageList = (List<Message>) query.execute();            
            if (messageList != null && !messageList.isEmpty()) {
                message = messageList.get(0);
                message.setStatus(MessageStatus.DELIVERED);             
            }

            transaction.commit();
        } finally {
            if (transaction != null && transaction.isActive()) {
                transaction.rollback();
            }

            pm.close();
        }
        return message;
    }

Answer 1

您似乎对来自设备的消息进行了长时间轮询。如果是这种情况，那么使用任务队列就没有任何实际价值。队列更多用于执行作业/业务处理。

您应该只使用端点并使用您拥有的最后一条消息的时间戳调用服务器。然后，每个设备将从数据存储中的查询中检索完全相同的结果集。 Memcache 将帮助解决配额问题。

您可以参考 Dropbox 的做法，他们使用发送的 delta 令牌来接收一组任何新的文件夹/文件更改。