过度声明跟踪令牌

编程问答 2022-05-14

问题描述

在将微服务扩展到 2 个副本时，我们注意到 TrackingEventProcessor 类的日志记录过多：

我们的 Axon 设置：

  • 轴突 3.4.3 版
  • spring-boot 版本 2.1.6.RELEASE
  • 作为 TokenStore 的内部 couchbase 实现，即 CouchBasetokenStore
  • Postgresql v11.7 作为事件存储
  • 每个跟踪事件处理器的段数 = 1
  • 跟踪事件处理器已配置 forSingleThreadedProcessing

我们经常看到以下消息：

Segment is owned by another node. Releasing thread to process another segment...
No Worker Launcher active. Using current thread to assign segments.
Using current Thread for last segment worker: TrackingSegmentWorker{processor=core_policy-view,segment=Segment[0/0]}
Fetched token: GapAwareTrackingToken{index=2369007,gaps=[]} for segment: Segment[0/0]

以下是名为 core_policy-view 的单个处理组的示例：

enter image description here

这是来自我们的登台环境。我们在我们的生产环境中没有看到相同的行为，即运行相同微服务的更多副本（5 个而不是 2 个），尽管是以前的版本。过多的日志记录令人担忧，因为它会导致以下查询也针对事件存储过度执行：

SELECT min ( globalIndex ) FROM DomainEventEntry WHERE globalIndex > ?

这是当微服务扩展到 2 个副本时它如何影响 APM 的图表：

enter image description here

通过查看 TrackingEventProcessor 类中的代码，我了解到 TrackingEventProcessor 负责将段分配给正在运行的实例。如果该段已被声明，则特定实例不应再打扰太多。然而，这似乎并没有发生。有时也可能发生的是两个节点之间的段/令牌乒乓。

目前我什至可能没有正确表达这个问题，我也不确定要问什么问题。如果有人能阐明我们是否做错了什么，我将不胜感激。我知道我们使用的 axon 版本很旧，我们在升级路线图上有它，但现在我们需要将系统的当前更改放入生产中，以便我们可以继续前进并在此之后开始升级.

编辑 1

以下是 CouchBasetokenStore 中负责声明令牌的方法：

@Override
public void storetoken(TrackingToken token,String processorName,int segment) throws UnabletoClaimTokenException {
    // Consider tracking CAS to avoid re-read and harness the CAS optimistic concurrency better.
    // Unfortunately,GapAwareTrackingToken is expected by RDBMS storage engines and can't be extended.
    JsonDocument doc = readOrCreateDocument(processorName,segment);
    if (GapAwareTrackingToken.class.isAssignableFrom(token.getClass())) {
        writeGapAwaretoken((GapAwareTrackingToken) token,doc);
    } else {
        writeTrackingToken(token,doc);
    }
    this.axonStateBucket.upsert(doc);
}

@Override
public TrackingToken fetchToken(String processorName,int segment) throws UnabletoClaimTokenException {
    JsonDocument doc = readOrCreateDocument(processorName,segment);
    String tokenClass = doc.content().getString(TOKEN_CLASS_FIELD);
    if (tokenClass == null) {
        return readGapAwaretoken(doc);
    } else {
        return readTrackingToken(doc);
    }
}

private JsonDocument readOrCreateDocument(String processorName,int segment) throws UnabletoClaimTokenException {
    String docId = getId(processorName,segment);
    JsonDocument doc = this.axonStateBucket.get(docId);
    if (doc == null) {
        try {
            doc = createDocument(processorName,segment);
        } catch (DocumentAlreadyExistsException e) {
            // Another instance beat us to it,read new token
            // which will most likely not be claimable.
            doc = this.axonStateBucket.get(docId);
        }
    }
    claimToken(doc);
    return doc;
}

private JsonDocument createDocument(String processorName,int segment) throws DocumentAlreadyExistsException {
    JsonObject content = JsonObject.create()
        .put(PROCESSOR_NAME_FIELD,processorName)
        .put(SEGMENT_FIELD,segment)
        .put(TYPE_FIELD,TOKEN_TYPE)
        .put(CLaim_EXPIRY_FIELD,formatInstant(Instant.Now().plus(claimDuration)))
        .put(OWNER_FIELD,nodeName);
    JsonDocument doc = JsonDocument.create(getId(processorName,segment),content);
    return this.axonStateBucket.insert(doc);
}

private void claimToken(JsonDocument document) throws UnabletoClaimTokenException {
    String originalOwner = document.content().getString(OWNER_FIELD);
    Instant originalClaimExpiry = DateTimeUtils.parseInstant(document.content().getString(CLaim_EXPIRY_FIELD));
    document.content()
        .put(CLaim_EXPIRY_FIELD,nodeName);
    if (nodeName.equals(originalOwner)) return;
    if ((originalClaimExpiry).isAfter(clock.instant())) {
        throw new UnabletoClaimTokenException(String.format("Claim for owner %s is still valid.",originalOwner));
    }
}

解决方法

我在 Allard 的帮助下设法解决了这个问题（请参阅问题评论）。修复方法是在 fetch() 方法中声明令牌后也保留令牌。我们还开始使用 Couchbase SDK 提供的 replace() 方法而不是 upsert() 方法，以更好地利用 CAS（比较和交换）乐观并发：

@Override
public void storeToken(TrackingToken token,String processorName,int segment) throws UnableToClaimTokenException {
    JsonDocument doc = readOrCreateDocument(processorName,segment);
    if (GapAwareTrackingToken.class.isAssignableFrom(token.getClass())) {
        writeGapAwareToken((GapAwareTrackingToken) token,doc);
    } else {
        writeTrackingToken(token,doc);
    }
    axonStateBucket.replace(doc);
}

@Override
public TrackingToken fetchToken(String processorName,segment);

    axonStateBucket.replace(doc); // readOrCreateDocument method claims,so we need to persist that

    String tokenClass = doc.content().getString(TOKEN_CLASS_FIELD);

    if (tokenClass == null) {
        return readGapAwareToken(doc);
    } else {
        return readTrackingToken(doc);
    }
}

其余代码与问题中的代码块相同。

axonaxon-framework