在ConsumerOffsetManager进行消费位点提交的时候如果在这个情况下有高并发会有一定概率丢失其中的一个队列的消费位点吗？

[Bngel]

ConsumerOffsetManager.java

private void commitOffset(final String clientHost, final String key, final int queueId, final long offset) {
    ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);
    if (null == map) {
        map = new ConcurrentHashMap<Integer, Long>(32);
        map.put(queueId, offset);
        this.offsetTable.put(key, map);
    } else {
        Long storeOffset = map.put(queueId, offset);
        if (storeOffset != null && offset < storeOffset) {
            LOG.warn("[NOTIFYME]update consumer offset less than store. clientHost={}, key={}, queueId={}, requestOffset={}, storeOffset={}", clientHost, key, queueId, offset, storeOffset);
        }
    }
    if (versionChangeCounter.incrementAndGet() % brokerController.getBrokerConfig().getConsumerOffsetUpdateVersionStep() == 0) {
        long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
        dataVersion.nextVersion(stateMachineVersion);
    }
}

在这段代码里，假设两个请求来自两个监听同一个topic@group的队列，那么这个时候假设一个队列先 ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);，这个时候 null == map成立，进行对应put，另一个队列同理，这样的话是不是覆盖掉了其中的一个offset？是不是因为只需要对offset的大小进行比较，如果较小的被覆盖掉也无所谓，所以这个地方并不需要保证强一致性？但是如果这样的话，是不是有可能出现极端情况下的消息重新消费？假如这个丢失了之后传了相同的消费位点过来也会判断是可以消费的？

感觉有点绕进去了，求解答。

[wangjinjing1]

看了代码，因为这个地方没有加锁，如果第一个请求走到map.put(queueId, offset)时候，又一个进来确实会有这个问题，这样必须在消费端校验是否消费过了，不然确实会重复消费

[Bngel]

感谢回复，😄
我理解在超高并发的时候还是会有这个问题的，那么rocketMQ为什么没有解决这个问题呢？是出于什么考虑呢？
目前我理解简单的方法的话可以需要通过一个读写锁来进行限制，那么加锁的开销和这个重复消费的问题应该如何权衡呢。

[wangjinjing1]

加锁确实能解决问题，如果消费时处理的业务、问题很多，那就会很慢，但是会拖慢整个业务链的时间。在高并发的情况下本身就要求系统响应快，这样可能会导致消息堆积，像重复消费这种问题，最简单解决方法就是消费前校验下，是否已经成功处理过了，如果处理过了，直接跳过该条消息

[Bngel]

那我理解了，就是在性能和重复消费中选择了性能，将重复消费的风险交给消费端来进行幂等处理。
感谢大佬解惑！ Thanks♪(･ω･)ﾉ

[3424672656]

正常情况下并不会发生这种问题，push消费时客户端负载均衡策略会保证每一个队列同时只被一个客户端消费