NSQ重新排队半个消息

Question

我有一个int64类型的Ids数组，这是我试图发布的Nsq消息。

nsqMsg := st{
    Action  :     "insert",
    Ids     :     Ids
    GID     :     Gids
}

msg, err := json.Marshal(nsqMsg)
if err != nil {
    log.Println(err)
    return err
}
err = nsqProducer.Publish(TOPIC-NAME, msg)
if err != nil {
    log.Println(err)
    return err
}

在我的消费者中，我一个接一个地获取每个Id，并根据我的Id从我的数据存储中获取信息。

因此，在获取时，可能会出现CreateObject方法返回错误的情况，因此我通过重新排队消息(给出错误的消息)来处理这种情况，因此可以重试。

for i := 0; i < len(data.Ids); i++ {

    Object, err := X.CreateObject(data.Ids[i)
        if err != nil {
            requeueMsgData = append(requeueMsgData, data.Ids[i])
            continue
        }
        DataList = append(DataList, Object)
    }

    if len(requeueMsgData) > 0 {
        msg, err := json.Marshal(requeueMsgData)
        if err != nil {
            log.Println(err)
            return err
        }
        message.Body = msg
        message.Requeue(60 * time.Second)
        log.Println("error while creating Object", err)
        return n
}

那么，这是做这件事的正确方式吗？这个案子有什么缺点吗？是不是再发布一次更好？

Answer 1

一些队列(如Kafka)支持确认，其中出队的项目直到消费者实际确认成功接收项目后才会从队列中删除。

此模型的优点是，如果消费者在消费之后但在确认之前死亡，则该项目将自动重新排队。你的模型的缺点是，在这种情况下，物品可能会丢失。

确认模型的风险在于，项目现在可能会被重复消费。如果消费者尝试有副作用的消费(如递增计数器或改变数据库)，但没有确认，因此重试可能不会产生预期的结果。(请注意，阅读nsq文档时，即使不对数据重新排队，也不能保证会发生重试，因此您的代码很可能不得不对此采取防御措施)。

你应该研究一下“恰好只有一次”与““至多一次”处理，如果你想更深入地理解这一点。

通读nsq文档，它看起来不支持确认，所以如果您必须使用nsq，这可能是最好的选择。

Answer 2

与dolan所说的一样，你可能会遇到一些情况：

• 主邮件心跳/租用超时，您将再次收到所有ids (来自原始邮件)。NSQ提供任何单个消息超时且永远不会完成的semantics.
• Requeue (回退到主IDS)

因为CreateObjects可以(大多数def会:p)多次传递消息，所以为了处理这种情况，nsq可以/应该是idempotent。

此外，重发是一个重要的安全机制，只有在所有单独的is或确认创建或成功重新排队时，才能找到原始消息，这确保了不会丢失数据。

IMO您处理它的方式看起来非常好，但最重要的考虑因素IMO是在接收到重复消息的环境中处理正确性/数据完整性。

另一种选择是对Requeue进行批处理，以便它尝试生成失败in的单个输出消息，这可以在任何给定时间减少队列中的消息数量：

考虑一条具有3个ids的消息：

消息id3 : id1、id2、id3

id1创建成功，id2和id3失败：

该程序可以尝试所有操作，并使用id2、id3发出单个重新排队消息。

但这也是一种权衡。