pom.xml:
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.ultiwill</groupId> <artifactId>ultiwill-es7</artifactId> <version>1.0-SNAPSHOT</version> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.2</version> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> </plugins> </build> <properties> <logback.version>1.2.3</logback.version> <slf4j.version>1.7.26</slf4j.version> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.11</version> <scope>test</scope> </dependency> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> <version>7.8.0</version> </dependency> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-client</artifactId> <version>7.8.0</version> </dependency> <dependency> <groupId>org.elasticsearch</groupId> <artifactId>elasticsearch</artifactId> <version>7.8.0</version> </dependency> <!--日志--> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>${slf4j.version}</version> </dependency> <dependency> <groupId>ch.qos.logback</groupId> <artifactId>logback-core</artifactId> <version>${logback.version}</version> </dependency> <!--json--> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.32</version> </dependency> </dependencies> </project>
EsClientUtil:
package com.ultiwill.utils; import org.apache.http.HttpHost; import org.apache.http.auth.AuthScope; import org.apache.http.auth.UsernamePasswordCredentials; import org.apache.http.client.CredentialsProvider; import org.apache.http.impl.client.BasicCredentialsProvider; import org.apache.http.impl.nio.client.HttpAsyncClientBuilder; import org.elasticsearch.action.bulk.*; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestClient; import org.elasticsearch.client.RestClientBuilder; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.unit.ByteSizeUnit; import org.elasticsearch.common.unit.ByteSizeValue; import org.elasticsearch.common.unit.TimeValue; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.*; import java.util.concurrent.TimeUnit; /** * @author chong.zuo * @date 2020/8/3 17:18 */ public class EsClientUtil { private static final Logger logger = LoggerFactory.getLogger(EsClientUtil.class); /** * 每次都取client太耗时，大约需要2秒左右，所以只取一次，放在内存中，不关闭，一直用 */ private static RestHighLevelClient client; private static BulkProcessor bulkProcessor; /** * 组装ES的hosts * * @return */ private static HttpHost[] assembleESAddress() { HttpHost httpHost1 = new HttpHost("192.168.100.110", 9201, "http"); HttpHost httpHost2 = new HttpHost("192.168.100.110", 9202, "http"); List<HttpHost> list = new ArrayList<HttpHost>(); list.add(httpHost1); list.add(httpHost2); HttpHost[] ipHost = new HttpHost[list.size()]; HttpHost[] httpHosts = list.toArray(ipHost); return httpHosts; } /** * 获取client连接 */ public static RestHighLevelClient getClient() { if (client == null) { synchronized (EsClientUtil.class) { try { if (client == null) { /** 用户认证对象 */ final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); /** 设置账号密码 */ credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("elastic", "123456")); /** 创建rest client对象 */ RestClientBuilder builder = RestClient.builder(assembleESAddress()) .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() { @Override public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) { return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider); } }); client = new RestHighLevelClient(builder); } } catch (Exception e) { logger.error("EsClient创建失败...." + client, e); } } } return client; } /** * 关闭client连接 */ public static void closeClient() { if (client != null) { synchronized (EsClientUtil.class) { try { client.close(); logger.info("ES Client 关闭成功..."); } catch (Exception e) { logger.error("ES Client关闭失败...", e); } } } } /** * 单条保存 * * @param index * @param id * @param m */ private static void saveData(String index, String id, Map<String, Object> m) { try { RestHighLevelClient client = getClient(); IndexRequest indexRequest = new IndexRequest(index) .id(id) .source(m); client.index(indexRequest, RequestOptions.DEFAULT); } catch (Exception e) { e.printStackTrace(); } } /** * 获取单例 BulkProcessor 批量处理类 */ public static BulkProcessor getBulkProcessor() { if (bulkProcessor == null) { synchronized (EsClientUtil.class) { try { if (bulkProcessor == null) { bulkProcessor = bulkProcessor(getClient()); } } catch (Exception e) { logger.error("BulkProcessor创建失败...." + bulkProcessor, e); } } } return bulkProcessor; } /** * 实例化 BulkProcessor * * @param client * @return */ public static BulkProcessor bulkProcessor(RestHighLevelClient client) { BulkProcessor.Listener listener = new BulkProcessor.Listener() { @Override public void beforeBulk(long executionId, BulkRequest request) { //bulk请求前执行 int numberOfActions = request.numberOfActions(); logger.info("ES Executing bulk [{}] with {} request ", executionId, numberOfActions); } @Override public void afterBulk(long executionId, BulkRequest request, BulkResponse response) { //bulk请求后执行 if (response.hasFailures()) { logger.error("ES Bulk [{}] executed with failures ", +executionId); } else { logger.info("ES Bulk [{}] completed in {} milliseconds ", executionId, response.getTook().getMillis()); } } @Override public void afterBulk(long executionId, BulkRequest request, Throwable failure) { // 失败后执行 logger.error("ES Bulk Failed to execute bulk ", failure); } }; BulkProcessor bulkProcessor = BulkProcessor.builder( (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener) // 达到刷新的条数 .setBulkActions(20000) // 达到 刷新的大小 .setBulkSize(new ByteSizeValue(10, ByteSizeUnit.MB)) // 固定刷新的时间频率 .setFlushInterval(TimeValue.timeValueSeconds(300)) //并发线程数 .setConcurrentRequests(5) // 重试补偿策略 .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3)).build(); return bulkProcessor; } public static void main(String[] args) { Date d = new Date(); String id = d.getTime() + ""; id = "12356789"; Map<String, Object> m = new HashMap<String, Object>(16); m.put("id", id); m.put("area_id", 1); m.put("camera_id", 1); m.put("log_time", new Date().toString()); m.put("age", 1); EsClientUtil.saveData("global_house_list", id, m); EsClientUtil.closeClient(); /*BulkProcessor bulkProcessor = EsClientUtil.getBulkProcessor(); IndexRequest one = new IndexRequest("posts").id("1").source(m); bulkProcessor.add(one);*/ } }
saveToEsBulk:
/** * 使用bulkProcessor 批量写数据入es * */ def save2ESByBulkProcessor(messages:util.ArrayList[util.Map[String,Any]],taskName:String):Unit={ if(messages.nonEmpty){ val bulk = IceBockEsConfig.getBulkProcessor(taskName) for(message <- messages){ val indexRequest = new IndexRequest() indexRequest.index(message.remove("indexName").toString).`type`(message.remove("indexType").toString).opType(DocWriteRequest.OpType.INDEX) if (message.containsKey("indexId")){ indexRequest.id(message.remove("indexId").toString) } indexRequest.source(message) bulk.add(indexRequest) } } }