spring integrate es.png
最近一直在用elasticsearch做一些数据的存储和查询 ，开始为了方便就直接使用了spring data elasticsearch ，期间也经历了很多坑，比如findByName(String name)这种接口默认只返回10条数据，自己也对源码做个跟踪，也研究了一下为什么elasticsearch要这样做。接下来我们通过分析spring data es的两个接口，来看一下是如果实现对es进行操作的。
1.findByName接口分析
- spring data es repository 接口
public interface UserRepository extends ElasticsearchRepository<User,String> {
/**
* 默认只返回10条
*/
List<User> findByName(String name);
List<User> deleteByName(String name);
}
- spring 启动分析构造UserRepository代理接口
启动的时候ElasticsearchRepositoryFactoryBean的方法会进行对应代理类的生成,并注入到spring容器
@Override
public void afterPropertiesSet() {
super.afterPropertiesSet();
Assert.notNull(operations, "ElasticsearchOperations must be configured!");
}
这里的super指是RepositoryFactoryBeanSupport，spring data jpa启动整体代理接口注入到spring容器中也是通过这个类做的，其中spring data jpa 对应的类是JpaRepositoryFactoryBean。
#其中RepositoryFactoryBeanSupport类中的下面这段代码比较核心
public void afterPropertiesSet() {
#getRepositoryMetadata是获取这个接口的元数据，就是这个接口的一些基本的属
this.repositoryMetadata = this.factory.getRepositoryMetadata(repositoryInterface);
#这个方法是生成这个接口的实现类
this.repository = Lazy.of(() -> this.factory.getRepository(repositoryInterface, repositoryFragmentsToUse));
#如果不是懒加载立即初始化
if (!lazyInit) {
this.repository.get();
}
}
接着我们分析上面的核心代码getRepository这个方法
public <T> T getRepository(Class<T> repositoryInterface, RepositoryFragments fragments) {
Assert.notNull(repositoryInterface, "Repository interface must not be null!");
Assert.notNull(fragments, "RepositoryFragments must not be null!");
RepositoryMetadata metadata = getRepositoryMetadata(repositoryInterface);
RepositoryComposition composition = getRepositoryComposition(metadata, fragments);
RepositoryInformation information = getRepositoryInformation(metadata, composition);
validate(information, composition);
//这里是构造一个对象，然后填充一些这个接口对应的entity信息，比如索引名，类型名等
Object target = getTargetRepository(information);
// Create proxy 接下来通过代理工厂创建代理
ProxyFactory result = new ProxyFactory();
result.setTarget(target);
result.setInterfaces(repositoryInterface, Repository.class, TransactionalProxy.class);
result.addAdvice(SurroundingTransactionDetectorMethodInterceptor.INSTANCE);
result.addAdvisor(ExposeInvocationInterceptor.ADVISOR);
postProcessors.forEach(processor -> processor.postProcess(result, information));
result.addAdvice(new DefaultMethodInvokingMethodInterceptor());
//这个DefaultMethodInvokingMethodInterceptor
//会为所有我们在repository接口中自定义的方法加上切面
result.addAdvice(new QueryExecutorMethodInterceptor(information));
composition = composition.append(RepositoryFragment.implemented(target));
result.addAdvice(new ImplementationMethodExecutionInterceptor(composition));
//生产代理类返回注入到spring容器
return (T) result.getProxy(classLoader);
}
接下来我们看看QueryExecutorMethodInterceptor类，主要是resolveQuery这个方法
this.queries = lookupStrategy.map(it -> {
SpelAwareProxyProjectionFactory factory = new SpelAwareProxyProjectionFactory();
factory.setBeanClassLoader(classLoader);
factory.setBeanFactory(beanFactory);
return repositoryInformation.getQueryMethods().stream()//
.map(method -> Pair.of(method, it.resolveQuery(method, repositoryInformation, factory, namedQueries)))//
.peek(pair -> invokeListeners(pair.getSecond()))//
.collect(Pair.toMap());
}).orElse(Collections.emptyMap());
resolveQuery 中会为每一个方法创建一个ElasticsearchPartQuery,其中ElasticsearchPartQuery是自定义操作实现的核心类,我们看一下代码
//构建类的时候会创建一个PartTree主要用来描述这个方法是什么方法，有没有分页等到
private final PartTree tree;
private final MappingContext<?, ElasticsearchPersistentProperty> mappingContext;
public ElasticsearchPartQuery(ElasticsearchQueryMethod method, ElasticsearchOperations elasticsearchOperations) {
super(method, elasticsearchOperations);
this.tree = new PartTree(method.getName(), method.getEntityInformation().getJavaType());
this.mappingContext = elasticsearchOperations.getElasticsearchConverter().getMappingContext();
}
//这个是在方法运行时调用时通过切面到这里然后转换成elasticsearch的查询接口
@Override
public Object execute(Object[] parameters) {
ParametersParameterAccessor accessor = new ParametersParameterAccessor(queryMethod.getParameters(), parameters);
CriteriaQuery query = createQuery(accessor);
if(tree.isDelete()) {
Object result = countOrGetDocumentsForDelete(query, accessor);
elasticsearchOperations.delete(query, queryMethod.getEntityInformation().getJavaType());
return result;
} else if (queryMethod.isPageQuery()) {
query.setPageable(accessor.getPageable());
return elasticsearchOperations.queryForPage(query, queryMethod.getEntityInformation().getJavaType());
} else if (queryMethod.isStreamQuery()) {
Class<?> entityType = queryMethod.getEntityInformation().getJavaType();
if (query.getPageable().isUnpaged()) {
int itemCount = (int) elasticsearchOperations.count(query, queryMethod.getEntityInformation().getJavaType());
query.setPageable(PageRequest.of(0, Math.max(1, itemCount)));
}
return StreamUtils.createStreamFromIterator((CloseableIterator<Object>) elasticsearchOperations.stream(query, entityType));
} else if (queryMethod.isCollectionQuery()) {
if (accessor.getPageable() == null) {
int itemCount = (int) elasticsearchOperations.count(query, queryMethod.getEntityInformation().getJavaType());
query.setPageable(PageRequest.of(0, Math.max(1, itemCount)));
} else {
query.setPageable(accessor.getPageable());
}
return elasticsearchOperations.queryForList(query, queryMethod.getEntityInformation().getJavaType());
} else if (tree.isCountProjection()) {
return elasticsearchOperations.count(query, queryMethod.getEntityInformation().getJavaType());
}
return elasticsearchOperations.queryForObject(query, queryMethod.getEntityInformation().getJavaType());
}
可以看到上面那个execute方法中主要有个ElasticsearchOperations类，负责调用es的方法,因为我们的List<User> findByName(String name)接口返回的是一个List，如果我们没有设置分页接口，这里会填充Unpaged.INSTANCE也就是表示客户端没有设置分页，es服务端会有默认填充。
image.png
然后逻辑走到了
elasticsearchOperations.queryForList这个方法，这个方法调用的是queryForPage方法。这样整个流程我们就走通了，我们也看到了为什么我们写的es repository接口默认返回的是10条数据，所以如果想用这个接口去做大批量数据查询的话是会出现问题的，需要自己实现原生的接口，用scroll方式去拉取数据。其中deleteByName方法用按同样方式分析。
2.userRepository.save(User user)接口分析
public <S extends T> S save(S entity) {
Assert.notNull(entity, "Cannot save 'null' entity.");
elasticsearchOperations.index(createIndexQuery(entity));
elasticsearchOperations.refresh(entityInformation.getIndexName());
return entity;
}
/**
* Index an object. Will do save or update
*
* @param query
* @return returns the document id
*/
String index(IndexQuery query);
@Override
public String index(IndexQuery query) {
String documentId = prepareIndex(query).execute().actionGet().getId();
// We should call this because we are not going through a mapper.
if (query.getObject() != null) {
setPersistentEntityId(query.getObject(), documentId);
}
return documentId;
}
/**
* refresh the index
*
* @param indexName
*
*/
void refresh(String indexName);
@Override
public void refresh(String indexName) {
Assert.notNull(indexName, "No index defined for refresh()");
client.admin().indices().refresh(refreshRequest(indexName)).actionGet();
}
可以看到在save entity的时候做了两件事情，首先是index，也就是这条数据会存到es，但是此时我们查询是不能查到的，因为查询到es数据需要建立对应的倒排索引，index只是把数据放到es中的buffer中，还没有建立倒排索引所以index后是不能立即搜索到es中对应的数据的，然后又调用了refresh方法，这个方法会触发es去清空buffer 并写入文件系统缓存，也就是会为我们的索引文件建立倒排索引，此时就可以搜索到了。所以save这个接口性能是很差的，比起mysql单挑插入，我这边测试了，性能下降一半，大概是300ms，服务器上面可能性能好一点，速度快一点；es插入最好调用批量的bulk接口
public <S extends T> List<S> save(List<S> entities) {
Assert.notNull(entities, "Cannot insert 'null' as a List.");
Assert.notEmpty(entities, "Cannot insert empty List.");
List<IndexQuery> queries = new ArrayList<>();
for (S s : entities) {
queries.add(createIndexQuery(s));
}
elasticsearchOperations.bulkIndex(queries);
elasticsearchOperations.refresh(entityInformation.getIndexName());
return entities;
}`
可以看到这里批量插入最后只调1次refresh，refresh操作会清空es的buffer，并且写入到文件系统缓存，这个操作开销还是比较大的。
网友评论