optimize: 优化第三方同步在多线程下同步出现重复条目的情况

CHANGELOG.MD

@@ -2,6 +2,10 @@
 
 更新日志文档，版本顺序从新到旧，最新版本在最前(上)面。
 
+# 0.17.7
+
+- 优化第三方同步在多线程下同步出现重复条目的情况
+
 # 0.17.6
 
 - 优化查询条目剧集数的逻辑

gradle.properties

@@ -1 +1 @@
-version=0.17.6
+version=0.17.7

server/src/main/java/run/ikaros/server/core/subject/service/impl/SubjectSyncPlatformServiceImpl.java

@@ -5,6 +5,8 @@
 import jakarta.annotation.Nullable;
 import jakarta.validation.constraints.NotNull;
 import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
@@ -36,6 +38,7 @@ public class SubjectSyncPlatformServiceImpl implements SubjectSyncPlatformServic
     private final SubjectService subjectService;
     private ApplicationContext applicationContext;
     private final SubjectSyncRepository subjectSyncRepository;
+    private final Set<String> pullingPlatformIdSet = new CopyOnWriteArraySet<>();
 
     /**
      * Construct.
@@ -219,6 +222,10 @@ public Mono<SubjectSync> findBySubjectIdAndPlatformAndPlatformId(Long subjectId,
     private Mono<Subject> syncBySubjectSynchronizer(@Nullable Long subjectId,
                                                     SubjectSyncPlatform platform,
                                                     String platformId) {
+        if (pullingPlatformIdSet.contains(platformId)) {
+            return Mono.empty();
+        }
+        pullingPlatformIdSet.add(platformId);
         return Flux.fromStream(extensionComponentsFinder.getExtensions(SubjectSynchronizer.class)
                 .stream())
             .filter(subjectSynchronizer -> platform.equals(subjectSynchronizer.getSyncPlatform()))
@@ -228,6 +235,7 @@ private Mono<Subject> syncBySubjectSynchronizer(@Nullable Long subjectId,
                 "No found available subject platform synchronizer for platform-id: "
                     + platform.name() + "-" + platformId)))
             .map(subjectSynchronizes -> subjectSynchronizes.get(0))
+            .subscribeOn(Schedulers.boundedElastic())
             .flatMap(subjectSynchronizer -> subjectSynchronizer.pull(platformId))
             .onErrorResume(Exception.class, e -> {
                 String msg =
@@ -244,7 +252,7 @@ private Mono<Subject> syncBySubjectSynchronizer(@Nullable Long subjectId,
                     .map(sub -> sub.setId(null)).flatMap(subjectService::create))
                 : subjectService.update(subject)
                 .then(Mono.defer(() -> subjectService.findById(subjectId))))
-            .subscribeOn(Schedulers.boundedElastic());
+            .doFinally(signalType -> pullingPlatformIdSet.remove(platformId));
     }