mysql转ElasticSearch的案例分析，mysqlelasticsearch

前言

最近工作中在进行一些技术优化，为了减少对数据库的压力，对于只读操作，在程序与db之间加了一层-ElasticSearch。具体实现是db与es通过bin-log进行同步，保证数据一致性，代码调用es查询数据，与mysql解耦。
优势：

• 减少与mysql的耦合，查询不依赖于mysql特性。因为当前ElasticSearch的势头如同json一样，轻量、简洁。
• ElasticSearch扩展性强，可以使用廉价机器平行扩展性能。
• ElasticSearch对所有字段进行了索引，不用在原mysql表中大量添加索引，减少了数据复杂度。

API的个人理解

由于es的java api文档不多，因此参照es官方文档的概念，自己总结了一些api的用法，个人看法，不保证正确。
ElasticSearch官方文档
Spring-data-es官方文档

迁移案例分析

实体

@Document(indexName = "dbName", type = "tableName", shards = 6)
public class UserInfo{
    /**
     * 主键
     */
    @JsonProperty("id")
    private Long id;

    /**
     * 用户编号
     */
    @JsonProperty("user_id")
    private String userId;

    /**
     * 分数
     */
    @JsonProperty("score")
    private String score;
    /**
     * 创建时间
     */
    @JsonProperty("order_time")
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    @Field(type = FieldType.Date, index = FieldIndex.not_analyzed, format = DateFormat.custom, pattern = "yyyy-MM-dd HH:mm:ss")
    private Date createTime;

• queryOne

select * from user_info where id = #{id}

public UserInfo getById(String id){
    CriteriaQuery query = new CriteraQuery(Criteria.where("id").is(id));
    UserInfo userInfo = elasticsearchTemplate.queryForObject(query, UserInfo.class);
}

• queryForList(小数据量)

select * from user_info where user_id in #{userIdList}

public List<UserInfo> getByUserIdList(List<String> userIdList){
    SearchQuery searchQuery = new NativeSearchQueryBuilder().
    withIndices(EsQueryConstant.obtainIndicesName("dbName","tableName")).
    withFilter(QueryBuilders.termsQuery("user_id",userIdList)).
    return elasticsearchTemplate.queryForList(searchQuery,UserInfo.class);
}

• queryForList(大数据量)

select * from user_info where crete_time > #{createTime}

public List<UserInfo> getByUserIdList(Date createTime){
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery().
filter(QueryBuilders.rangeQuery("create_time").gt(new DateTime(createTime).toString("yyyy-MM-dd HH:mm:ss")));
    SearchQuery searchQuery = new NativeSearchQueryBuilder().
    withIndices(EsQueryConstant.obtainIndicesName("dbName","tableName")).
    withFilter(boolQueryBuilder ).build();
String scrollId = elasticsearchTemplate.scan(searchQuery, TimeValue.timeValueMinutes(20).getMillis(), false);
        List<UserInfo> result= Lists.newArrayList();
        while (true) {
            Page<UserInfo> userInfoPage = elasticsearchTemplate.scroll(scrollId, TimeValue.timeValueMinutes(20).getMillis(), UserInfo.class);
            List<UserInfo> userInfoContent= orderEsPage.getContent();
            if (CollectionUtils.isEmpty(userInfoContent)) {
                break;
            }
            result.addAll(userInfoContent);
        }
        return result;
}

由于es采用的是分布式存储，所以在数据量大到一定程度的情况下，分页已经变得不可行。比如要拿1000-1010的数据，假设es有6个分片，则每个分片都要拿到1010条数据，总体排序以后取到1000-1010的数据。这样的计算显然是不可能的。所以如果数据量够大，应当使用游标的方式查询数据。虽然指定了页大小，但是这只针对于每一片，实际得到的数据不超过片数*页大小。一直循环，直到所有分片都没有满足条件的数据为止。

• queryForPage

select * from user_info where score != #{score} limit #{pageIndex},#{pageSize}

public Page<UserInfo> getByUserIdList(String score,int pageIndex,int pageSize){
BoolQueryBuilder query= QueryBuilders.boolQuery().
mustNot(QueryBuilders.termQuery("score", score);
    SearchQuery searchQuery = new NativeSearchQueryBuilder().
    withIndices(EsQueryConstant.obtainIndicesName("dbName","tableName")).
    withPageable(new PageRequest(pageIndex, pageSize)).
    withFilter(query).build();
    return elasticsearchTemplate.queryForPage(searchQuery,UserInfo.class);
}

后记

本文简单的介绍了mysql转ElasticSearch时的一些场景的案例，API并不难，只是相关资料少，很多功能只能探索前进，以后用到了更深入的功能会继续更新。