当前位置：首页 > article >正文

同步数据至ES时，数据丢失问题处理

article 2025/1/19 2:35:54

问题背景：

数据同步到es过程中，发现数据丢失问题，原因正是因为写入索引前会先删除索引导致！

总体流程：

使用别名索引E
redis获取当前索引B(即E指向B)，获取新索引A
初始化新索引A，将数据存储到新索引A
redis存储当前索引A，别名索引E指向A

代码实现：

redis获取当前索引B

    public Class<?> getCurrentIndexByAlias(@NonNull String alias, Class<?> defaultClass) {
        String value = stringRedisTemplate.boundValueOps(INDEX_ALIAS +"_"+ alias).get();
        if (value == null) {
            return defaultClass;
        }
        return Class.forName(value);
    }

获取新索引A,存储数据

     Method method = currentIndex.getMethod("other");
     //执行方法获取新索引
     Class<?> newIndex = (Class<?>) method.invoke(currentIndex.getDeclaredConstructor().newInstance());
     //初始化索引
     initIndex(newIndex);
     //存储数据
     ...........

redis存储当前索引A

	redisRepository.switch2IndexOfAlias("E", newIndex);
    public void switch2IndexOfAlias(String alias, Class<?> clazz) {
        stringRedisTemplate.boundValueOps(INDEX_ALIAS +"_"+ alias).set(clazz.getName());
    }

别名索引E指向A

 esRepository.switchIndex("E",newIndex.getAnnotation(Document.class).indexName(),
                    currentIndex.getAnnotation(Document.class).indexName());

    public <T> void switchIndex(String alias, String newIndexName, String oldIndexName) {
        AliasActions aliasActions = new AliasActions();
        //更新新索引
        if (existsIndex(newIndexName)) {
            AliasAction.Add addAction = new AliasAction.Add(AliasActionParameters.builder()
                    .withIndices(newIndexName).withAliases(alias).build());
            aliasActions.add(addAction);
        }
        //删除旧索引
        if (existsIndex(oldIndexName)) {
            AliasAction.Remove removeAction = new AliasAction.Remove(AliasActionParameters.builder()
                    .withIndices(oldIndexName).withAliases(alias).build());
            aliasActions.add(removeAction);
        }
        //绑定别名
        elasticsearchOperations.indexOps(IndexCoordinates.of(newIndexName)).alias(aliasActions);
    }