最近项目中使用ES进行数据存储，但是总是出现数据库的数据更新完成，同步到ES的过程延迟，根据断点调试查看Spring的Elasticsearch的操作api如下：
1、spring操作es的api// 构建文档和更新条件
Document document = Document.from(JsonUtil.buildNonNullMapperWithJavaTimeModule().fromObjectToMap(entity));
UpdateQuery updateQuery = UpdateQuery.builder((stringIdRepresentation(entity.getId()))).withDocument(document).withTimeout("60s").build();
// 更新ES数据
elasticsearchOperations.update(updateQuery, entityInformation.getIndexCoordinates());
2、elasticsearchOperations.update调用的底层
// ElasticsearchTemplate#update()
public UpdateResponse update(UpdateQuery query, IndexCoordinates index) {
UpdateRequestBuilder updateRequestBuilder = requestFactory.updateRequestBuilderFor(client, query, index);
org.elasticsearch.action.update.UpdateResponse updateResponse = updateRequestBuilder.execute().actionGet();
UpdateResponse.Result result = UpdateResponse.Result.valueOf(updateResponse.getResult().name());
return new UpdateResponse(result);
}
3、继续查看UpdateRequestBuilder
public UpdateRequestBuilder updateRequestBuilderFor(Client client, UpdateQuery query, IndexCoordinates index) {
UpdateRequestBuilder updateRequestBuilder = client.prepareUpdate(index.getIndexName(), IndexCoordinates.TYPE,
query.getId());
if (query.getScript() != null) {
Map params = query.getParams();
if (params == null) {
params = new HashMap<>();
}
Script script = new Script(ScriptType.INLINE, query.getLang(), query.getScript(), params);
updateRequestBuilder.setScript(script);
}
// 判断refresh是否配置
if (query.getRefresh() != null) {
updateRequestBuilder.setRefreshPolicy(query.getRefresh().name().toLowerCase());
}
return updateRequestBuilder;
}
4、RefreshPolicy 类
enum RefreshPolicy implements Writeable {
NONE("false"),
IMMEDIATE("true"),
WAIT_UNTIL("wait_for");
private final String value;
RefreshPolicy(String value) {
this.value = value;
}
public String getValue() {
return value;
}
public static RefreshPolicy parse(String value) {
for (RefreshPolicy policy : values()) {
if (policy.getValue().equals(value)) {
return policy;
}
}
if ("".equals(value)) {
// Empty string is IMMEDIATE because that makes "POST /test/test/1?refresh" perform
// a refresh which reads well and is what folks are used to.
return IMMEDIATE;
}
throw new IllegalArgumentException("Unknown value for refresh: [" + value + "].");
}
public static RefreshPolicy readFrom(StreamInput in) throws IOException {
return RefreshPolicy.values()[in.readByte()];
}
@Override
public void writeTo(StreamOutput out) throws IOException {
out.writeByte((byte) ordinal());
}
}
在UpdateQuery这个类中有这个属性：
private Refresh refresh;
所以我们再调用的入口处进行设置数据刷新策略，我们选的是UpdateQuery.Refresh.True策略，即在数据读取前保证一致。
UpdateQuery updateQuery = UpdateQuery.builder((stringIdRepresentation(entity.getId()))) .withRefresh(refresh) .withDocument(document) .withDocAsUpsert(true) .build();注意：
UpdateQuery.Refresh策略不支持批量更新操作，只能进行单个数据的更新设置。
-------------如果对你有用，请给个赞，谢谢~~
-------------欢迎各位留言交流，如有不正确的地方，请予以指正。【Q:981233589】
