有效处理kafka消息的顺序消费方法

在IT行业这个发展更新速度很快的行业，只有不停止的学习，才不会被行业所淘汰。如果你是Golang学习者，那么本文《有效处理kafka消息的顺序消费方法》就很适合你！本篇内容主要包括##content_title##，希望对大家的知识积累有所帮助，助力实战开发！

背景

生产者生产一些数据并按顺序发送到 kafka，例如：

{uuid: 123 状态: 1}

{uuid: 123 状态: 3}

状态 1 表示开始

状态 3 表示成功

我使用 sarama.newconsumergroup(xx, xx, config).consume(xx, xx, myhandler) 来使用代码：

func (h myhandler) consumeclaim(sess sarama.consumergroupsession, claim sarama.consumergroupclaim) error {
    for msg := range claim.messages() {

        key := fmt.sprintf("%q-%d-%d", msg.topic, msg.partition, msg.offset)
        _, err := rdb.redisclient.get(h.ctx, key).result()
        if err == redis.nil {
            msgqueue <- msg.value
            sess.markmessage(msg, "")
            rdb.redisclient.set(h.ctx, key, none, 12*time.hour)
        } else if err != nil {
            log.errorln("redisclient get key error : ", err)
            return err
        } else {
            continue
        }

    }
    return nil
}

msgqueue := make(chan interface{}, 1000)

然后我将 msgqueue 中的值解码为结构并将一条记录插入到 mysql 中。

问题

通常情况下，最终数据状态为“3”，但我发现有时为“1”

而且我发现通道 msgqueue 中的消息顺序不固定。

那么如何确保数据的最终状态为3？

如何修复

我提供了一个方法，但还不够好，看看如何优化它。

conn := &gorm.DB{}
    data := &Log{}
    if data.Status != 1 {
        conn = conn.Clauses(clause.OnConflict{
            Columns:   []clause.Column{{Name: "uuid"}},
            DoUpdates: clause.AssignmentColumns([]string{"status"}),
        })
    }
    conn.Create(data)
    return conn.Error

并且mysql对uuid有唯一约束索引。

当数据顺序为{uuid: 123 status: 1}时， {uuid: 123 status: 3}，没错。

当数据顺序为{uuid: 123 status: 3}时， {uuid: 123 status: 1}，最终状态也是对的，但是会返回错误error 1062: duplicate entry '123' for key 'unique_index_uuid'。

这并不美丽。那么我该如何优化或者还有其他方法吗？

正确答案

这取决于主题分区。 Kafka 不提供主题内的排序保证，仅提供分区内的排序保证。

换句话说，如果您发送消息 A，然后发送消息 B 到分区 0，那么顺序将是：先发送 A，然后发送 B。但是如果它们最终位于不同的分区上，则可能会发生先将 B 写入其分区，然后再将 A 写入其分区的情况。

这里引用了 Confluence 网站上的一段话：

Link

以上就是本文的全部内容了，是否有顺利帮助你解决问题？若是能给你带来学习上的帮助，请大家多多支持golang学习网！更多关于Golang的相关知识，也可关注golang学习网公众号。