1.3.2版本较为核心的类DataSyncer存在并发问题

[shizhengxing]

文件路径nacos/naming/src/main/java/com/alibaba/nacos/naming/consistency/ephemeral/distro/DataSyncer.java

submit方法中，第120行的代码片段1

Map<String, Datum> datumMap = dataStore.batchGet(keys);

这里面从dataStore取出了keys对应的数据，并在接下来132行将其序列化后，发送到其他naming节点。
132行代码

boolean success = NamingProxy.syncData(data, task.getTargetServer());

但是在发送成功前（执行121至132行间的逻辑时），taskMap中的keys依旧存在，发送成功后才会从taskMap移除，这里存在并发问题。
我们先看88行至105行的代码逻辑

if (task.getRetryCount() == 0) {
    Iterator<String> iterator = task.getKeys().iterator();
    while (iterator.hasNext()) {
        String key = iterator.next();
        if (StringUtils.isNotBlank(taskMap.putIfAbsent(buildKey(key, task.getTargetServer()), key))) {
            // associated key already exist:
            if (Loggers.DISTRO.isDebugEnabled()) {
                Loggers.DISTRO.debug("sync already in process, key: {}", key);
            }
            iterator.remove();
        }
    }
}
if (task.getKeys().isEmpty()) {
    // all keys are removed:
    return;
}

当其他线程执行到88行至105行间的代码时，如果相同的key存在于taskMap中，这个key将不会再执行后续的流程，会导致第一个线程执行132行同步到其他节点的数据是旧的

[shizhengxing]

如果确认的话，我可以来修复这个问题

[KomachiSion]

这里在很极限的情况下的确可能存在问题，但是这个Distro本身就是AP协议，还有对账机制可以修复遗漏的同步信息，详见/checksum接口。

您说打算修复这个问题，能先同步一下修复的方案吗？讨论下， 如果在不影响性能的情况下，能修复掉最好。

[shizhengxing]

此问题的原因是执行143行taskMap.remove操作时，无法知道keys对应的数据在前面是否有更新，但是为了确保相同的key同步到到其他naming节点的顺序性，不能简单通过在129行执行serialize前移除taskMap中的key来解决，我当前想到的办法是引入状态机制，记录序列化后又有更新这个状态

步骤1，将63行taskMap定义的value的数据结构改成

private Map<String, StatusKey> taskMap ......

，类型定义如下：

private static class StatusKey {
    //刚开始
    static int begin = 0;
    //已经被序列化，正在发送给其他naming节点
    static int already_serialized = 1;
    //已经被序列化，但是又有更新
    static int serialized_but_to_sync_agent = 2;
    //需要自旋等待put到taskMap，对于已完成数据发送的线程来说是将从taskMap中remove此key
    static int loop_or_to_remove = 3;
    
    String key;
    AtomicInteger status = new AtomicInteger(0);
}

处理流程图如下：
步骤2，92行的String key = iterator.next();改为StatusKey key =new StatusKey(iterator.next();
步骤3，假设线程1执行93行的taskMap.putIfAbsent操作返回值假定为R，R的类型为StatusKey

3.1) 如果R==null || R.status==0 || R.status==2,直接return；
3.2) 如果R.status==1，在R的status字段上执行CAS操作更改为2，CAS成功则return，否则状态将为3，此时自旋执行93行的taskMap.putIfAbsent
此处简述为

if(R.status==1) then CAS(1,2) -> if ok -> return
if(R.status==1) then CAS(1,2) -> if not ok（说明状态为3） -> loop line 93等待remove

3.3) 如果 R.status==3 -> loop line 93等待remove；

步骤4，假设线程2执行序列化以及发送到其他naming节点的步骤，也就是执行114到144行之间的代码

4.1) 在114行遍历keys，从taskMap执行get得到值X，X的类型为StatusKey
4.2) 对X的status字段执行CAS描述如下

CAS(0,1) -> send other node if not success -> set 0 -> 走原有的retrySync逻辑
CAS(0,1) -> send other node if success -> CAS(1,3) -> if ok -> remove key form taskMap
CAS(0,1) -> send other node if success -> CAS(1,3) -> if not ok(说明状态为2) -> 收集本次状态为2的key继续执行CAS(2,3) -> remove key form taskMap -> 收集到的key重新执行85行的submit方法

[shizhengxing]

这个方案唯一可能影响性能的是3.2步骤的自旋操作，但是只有在4.2步骤status=3的时候才会走这个逻辑，然而4.2识别到status=3时，只会迅速remove这个key，因此对性能几乎无影响，但是能很好的解决一致性的这个问题，个人认为整体是可行的，编码难度不大

[KomachiSion]

我觉得问题复杂了，近期我们计划把AP协议下沉，到时候这个地方会重新修改，改成类似两层key的机制，
第一层task key是用来做归并的，当需要同步的时候这个key就移除了，移除key之后再get数据，然后同步。此时如果get完数据又有变更，会进入第一层key去做归并。
同样 如果同步失败，也回滚回第一层key做归并。
这样同步数据和归并同步作业就不会互相影响。

[shizhengxing]

引入状态机制确实变得复杂了些，你提到的下层AP协议计划是在哪个版本呢？你的描述比较简单，还看不出来是如何确保同一个key同步的顺序性的

[KomachiSion]

Refer to #3642

[KomachiSion]

Distro refactor has done. the new design fixed this problem.