环境配置

SpringBoot-2.1.8 + Elasticsearch-6.2.2

版本问题

spring官网提供

• elasticsearch
• spring-data-elasticsearch
• spring-boot-starter-data-elasticsearch

有三种集成方式可选择，我这里的springboot2.1.8对elasticsearch最高支持到6.2.2

连接方式

• REST API，端口9200，这种客户端的连接方式是RESTful风格的，使用http的方式进行连接，分为Java Low Level REST Client和Java High Level REST Client。
• Transport连接端口9300，这种客户端连接方式是直接连接ES的节点(给定多个集群节点，将客户端负载均衡地向这个节点地址集发送请求)，使用TCP的方式进行连接。
• spring封装后的调用：spring-data-elasticsearch和spring-boot-starter-data-elasticsearch。

测试代码

import org.junit.Test;import org.junit.runner.RunWith;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;import org.springframework.data.elasticsearch.core.query.IndexQuery;import org.springframework.test.context.junit4.SpringRunner;import org.xinhua.cbcloud.pojo.DocLog;import org.xinhua.cbcloud.repository.DocLogRepository;import org.xinhua.cbcloud.util.IDUtil;import java.util.ArrayList;import java.util.LinkedList;import java.util.List;@RunWith(SpringRunner.class)@SpringBootTestpublic class LyfeifeiApplicationTests { private Logger logger = LoggerFactory.getLogger(LyfeifeiApplicationTests.class); @Autowired ElasticsearchTemplate elasticsearchTemplate; @Autowired DocLogRepository docLogRepository; @Test public void createIndex() { // 创建索引 System.out.println(elasticsearchTemplate.createIndex(DocLog.class)); } @Test public void deleteIndex() throws Exception { System.out.println(elasticsearchTemplate.deleteIndex(DocLog.class)); } @Test public void insert() throws Exception { int counter = 0; logger.info("开始执行"); List docLogs = new ArrayList<>(); for (int i = 0; i < 1000000; i++) { DocLog docLog = new DocLog(); docLog.setId(IDUtil.getId()); docLog.setDocId(String.valueOf(IDUtil.getId())); docLog.setMessageId("1569482824826#b807cc4a-6258-4f61-a445-c97006d9512c"); docLog.setContext("111111"); docLogs.add(docLog); } // 索引队列 List indexQueries = new LinkedList<>(); for (DocLog docLog : docLogs) { IndexQuery indexQuery = new IndexQuery(); indexQuery.setId(String.valueOf(IDUtil.getId())); indexQuery.setObject(docLog); indexQuery.setIndexName("doclog"); indexQuery.setType("docs"); indexQueries.add(indexQuery); if (counter % 5000 == 0) { elasticsearchTemplate.bulkIndex(indexQueries); indexQueries.clear(); } counter++; } if (indexQueries.size() > 0) { elasticsearchTemplate.bulkIndex(indexQueries); } logger.info("执行结束"); }}

测试结果

写在最后

目前的es为单节点并没有做集群，100w数据生成再导入es不到两分钟，感觉还能接受吧。这个本身没有太多难点，es的强大更表现在查询检索方面，bulkIndex是它提供的一个批量导入方法。es的数据插入小弟尝试过多线程批量导入，但是无论是单条还是批量都以失败告终，应该是其内部的锁造成的，暂时还没有做深入研究。