使用Spring Data Elasticsearch查询Elasticsearch新增数据的问题咨询

1. 仅拉取新增写入数据的实现方式 #

核心逻辑是使用游标标记上次同步的位置，每次查询只返回游标之后的新数据，有两种常用实现方案：

• 基于业务时间戳的游标方案
  给ES索引文档新增createTime字段，写入ES时自动填充当前系统时间，每次同步完成后记录本次拉取到的最大createTime值存储到本地/Redis/MySQL，下次查询只筛选createTime大于该值的文档即可。
  代码示例：
  实体类配置：

  @Document(indexName = "你的业务索引名")
  public class BizDocument {
      @Id
      private String id;
      // 其他业务字段省略
      @Field(type = FieldType.Date, format = DateFormat.date_time)
      private Date createTime;
      // 省略getter/setter
  }
  

  Repository 新增查询方法：

  public interface BizDocumentRepository extends ElasticsearchRepository<BizDocument, String> {
      List<BizDocument> findByCreateTimeGreaterThan(Date lastSyncTime);
  }
  

  同步逻辑：

  // 从存储中读取上次同步的最大时间，首次全量同步可传new Date(0)
  Date lastSyncTime = getLastSyncTime();
  List<BizDocument> newData = bizDocumentRepository.findByCreateTimeGreaterThan(lastSyncTime);
  // 处理新增数据逻辑省略
  // 更新同步游标
  if (!newData.isEmpty()) {
      Date maxCreateTime = newData.stream().map(BizDocument::getCreateTime).max(Date::compareTo).get();
      saveLastSyncTime(maxCreateTime);
  }
  

• 基于ES内置字段的游标方案
  该方案比时间戳更准确，不会出现同一时间写入导致的漏数据问题。ES内置_seq_no字段是文档级别的全局递增序列号，新增/修改文档时自动递增，不会重复，直接映射该字段作为游标即可。
  实体类配置：

  @Document(indexName = "你的业务索引名")
  public class BizDocument {
      @Id
      private String id;
      // 其他业务字段省略
      @SeqNo
      private Long seqNo;
      @PrimaryTerm
      private Long primaryTerm;
      // 省略getter/setter
  }
  

  Repository新增查询方法：

  List<BizDocument> findBySeqNoGreaterThan(Long lastSeqNo);
  

  同步逻辑和时间戳方案一致，每次存储上次拉取到的最大seqNo作为游标即可。如果数据量较大，建议搭配SearchAfter分页查询避免内存溢出。

2. 异步获取新增记录的实现方式 #

ES本身没有内置增量数据推送的触发器能力，可通过以下方案实现异步获取：

• 定时轮询方案
  实现最简单，无额外组件依赖，适合对实时性要求不高的场景。直接使用Spring自带的定时任务注解，每隔固定时间执行一次上述的增量查询逻辑即可。
  代码示例：

  @Component
  public class EsSyncTask {
      // 每隔5秒执行一次增量同步，可根据业务需求调整间隔
      @Scheduled(fixedDelay = 5000)
      public void syncIncrementalData() {
          // 执行上述增量拉取+数据处理逻辑
      }
  }
  

• 变更事件订阅方案
  如果是云厂商托管的ES服务，大部分都自带索引变更事件订阅能力，可直接配置将新增/更新事件推送到Kafka/RocketMQ等消息队列，你的处理应用直接消费消息队列即可拿到实时新增数据。
  如果是自建ES，可以通过Logstash配置定时增量查询，将查询到的新增数据输出到消息队列，下游应用消费消息即可。
• 业务旁路同步方案
  如果你有权限修改写入ES的业务代码，可在业务写入ES的逻辑中同时向消息队列发送一条数据变更消息，下游处理应用直接消费消息，该方案实时性最高，实现复杂度最低。

内容的提问来源于stack exchange，提问作者vr3w3c9