延迟确认 GCloud Pub/Sub 消息

Question

我正在尝试实现一个类似于 AWS 或 Azure 队列的 PubSub 客户端，但是，我遇到了 gcloud cpp sdk 的问题。

更新：删除了不必要的细节。

首先，提供的example 不能开箱即用——我必须在session.cancel() 之前休眠，否则无法确认消息。是否有可靠的方法等到 ack() 操作完成并检查其状态？至少我想确保服务器收到我的请求。

此外，c++ API 似乎只提供了一个不适合我的用例的异步方法。

我需要实现以下接口，该接口通过依赖注入插入到更大的系统中。该系统在其他云上的生产中工作，因此我无法更改架构。只需要实现接口即可。

template<typename TItem>
class Queue{
public:
    /*!
    * Dequeues message from the queue 
    * Returns true on success
    */
    virtual bool Dequeue( TItem & item) = 0;

    /*!
    * Discards(deletes) the item with from the cloud queue.
    */
    virtual void Discard(const TReceipt & receipt) = 0;
};

队列的实际实现将提供一个序列化器，该序列化器将 TItem 序列化为 JSON 并返回。

AWS 和 Azure SDK 为每个出队的消息提供一个收据，以便我以后可以丢弃它。 pubsub SDK 的收据是 AckHandler 对象，它绑定到会话。

一个明显错误的解决方案是保持会话打开并在 lambda 中等待另一个 condition_variable，直到下一次调用 Dequeue 方法。但是，这看起来像是一个快速而肮脏的解决方案。 使用 Pub/Sub 实现此功能的正确方法是什么？

Answer 1

是否有可靠的方法等到 ack() 操作完成并检查其状态？

并非如此，因为ack() 操作是尽力而为。即使您等待ack() 到达服务器，也不能保证消息不会被重新发送。是的，即使ack() 成功，服务也可能会重新发送消息。

请注意，库会自动延长您消息的租期以避免重新发送，直到您明确 ack() 或 nack()，因此 ack() 中的延迟不会影响正确性除非您关闭在ack()之后不久申请。

AWS 和 Azure SDK 为每个出队的消息提供一个收据，以便我以后可以丢弃它，但我不知道如何使用 pub/sub 来做到这一点。

您可以在AckHandler 上调用nack() 来丢弃消息，这意味着该消息将被重新发送到另一个实例。这就是你所说的“丢弃”吗？

使用 Pub/Sub 实现此功能的正确方法是什么？

嗯，我不确定我是否遵循。我可以猜测Dequeue() 和Discard() 的语义，如果我猜错了请见谅。无论如何，尚不清楚这是否真的适用于任何类型？喜欢TReceipt == int？

需要注意的是这里有很多猜测，你可以这样做：

class PubsubBuffer {
 private:
  std::mutex mu_;
  std::dequeue<std::pair<pubsub::Message, AckHandlerWrapper>> queue_;

 public:
  virtual bool Dequeue(pubsub::Message& item, AckHandlerWrapper& receipt) {
    std::unique_lock<std::mutex> lk(mu_);
    if (queue_.empty()) return false;
    auto& f = queue_.front();
    item = std::move(f.first);
    receipt = std::move(.second);
    queue_.pop_front();
    return true;
  }
  virtual void Discard(AckHandlerWrapper const& receipt) {
    std::move(receipt.ack_handler).ack();
  }
  void Push(pubsub::Message m, pubsub::AckHandler h) {
    std::unique_lock<std::mutex> lk(mu_);
    queue_.push_back(
      std::make_pair(std::move(m), AckHandlerWrapper(std::move(h)));
  }
};


std::shared_ptr<PubsubBuffer> F(pubsub::Subscriber s) {
  auto buffer = std::make_shared<PubsubBuffer>();
  auto handler = [buffer](pubsub::Message m, pubsbu::AckHandler h) {
    buffer.Push(std::move(m), std::move(h));
  };
  return buffer;
}