确实在高级消费者块上提交.它会发射武器?

Answer 1

在高级消费者块上提交订单，直到偏移成功地提交为止？

看起来commitOffsets()通过每个消费者循环，如果其偏移发生了变化，则调用updatePersistentPath，如果这样通过zkClient.writeData(path, getBytes(data))写入数据。看来，虽然commitOffsets() 确实块直到所有偏移都犯下为止。

这是commitOffsets()（参考）的源代码：

public void commitOffsets() {
    if (zkClient == null) {
        logger.error("zk client is null. Cannot commit offsets");
        return;
    }
    for (Entry<String, Pool<Partition, PartitionTopicInfo>> e : topicRegistry.entrySet()) {
        ZkGroupTopicDirs topicDirs = new ZkGroupTopicDirs(config.getGroupId(), e.getKey());
        for (PartitionTopicInfo info : e.getValue().values()) {
            final long lastChanged = info.getConsumedOffsetChanged().get();
            if (lastChanged == 0) {
                logger.trace("consume offset not changed");
                continue;
            }
            final long newOffset = info.getConsumedOffset();
            //path: /consumers/<group>/offsets/<topic>/<brokerid-partition>
            final String path = topicDirs.consumerOffsetDir + "/" + info.partition.getName();
            try {
                ZkUtils.updatePersistentPath(zkClient, path, "" + newOffset);
            } catch (Throwable t) {
                logger.warn("exception during commitOffsets, path=" + path + ",offset=" + newOffset, t);
            } finally {
                info.resetComsumedOffsetChanged(lastChanged);
                if (logger.isDebugEnabled()) {
                    logger.debug("Committed [" + path + "] for topic " + info);
                }
            }
        }
    }
}

和updatePersistentPath(...)（参考）：

public static void updatePersistentPath(ZkClient zkClient, String path, String data) {
    try {
        zkClient.writeData(path, getBytes(data));
    } catch (ZkNoNodeException e) {
        createParentPath(zkClient, path);
        try {
            zkClient.createPersistent(path, getBytes(data));
        } catch (ZkNodeExistsException e2) {
            zkClient.writeData(path, getBytes(data));
        }
    }
}