Java 类org.elasticsearch.client.RestHighLevelClient 实例源码

@Override
public void setupTest(BackendListenerContext context) throws Exception {
    try {
        this.index        = context.getParameter(ES_INDEX);
        this.bulkSize     = Integer.parseInt(context.getParameter(ES_BULK_SIZE));
        this.timeoutMs = JMeterUtils.getPropDefault(ES_TIMEOUT_MS, DEFAULT_TIMEOUT_MS);
        this.buildNumber  = (JMeterUtils.getProperty(ElasticsearchBackend.BUILD_NUMBER) != null 
                && JMeterUtils.getProperty(ElasticsearchBackend.BUILD_NUMBER).trim() != "") 
                ? Integer.parseInt(JMeterUtils.getProperty(ElasticsearchBackend.BUILD_NUMBER)) : 0;
        String host         = context.getParameter(ES_HOST);
        int port         = Integer.parseInt(context.getParameter(ES_PORT));
        this.client       = new RestHighLevelClient(
                    RestClient.builder(
                        new HttpHost(host, port, context.getParameter(ES_SCHEME, "http")))
                    .setRequestConfigCallback(requestConfigBuilder -> 
                         requestConfigBuilder
                                    .setConnectTimeout(5000)
                                    .setSocketTimeout((int)timeoutMs))
                    .setMaxRetryTimeoutMillis(60000));
        this.bulkRequest = new BulkRequest().timeout(TimeValue.timeValueMillis(timeoutMs));
        super.setupTest(context);
    } catch (Exception e) {
        throw new IllegalStateException("Unable to setup connectivity to ES", e);
    }
}

@Inject
public ESSearchService(@Named("artistDao") ArtistDao artistDao,
                       @Named("albumDao") MaprDbDao<Album> albumDao) {

    this.artistDao = artistDao;
    this.albumDao = albumDao;

    RestClient lowLevelRestClient = RestClient.builder(new HttpHost(ES_REST_HOST, ES_REST_PORT, "http")).build();
    this.client = new RestHighLevelClient(lowLevelRestClient);
}

@Override
public void startComponent() {
  if (lowLevelClient == null) {
    // Create a connection for each contact point.
    Set<HttpHost> hosts = contactPoints.stream()
            .map(s -> new HttpHost(s, port))
            .collect(Collectors.toSet());
    // Initialize the low-level REST client managing the underlying connections to ElasticSearch.
    lowLevelClient = RestClient.builder(hosts.toArray(new HttpHost[contactPoints.size()])).build();
    // Initialize the high-level REST client which sends the actual requests to ElasticSearch.
    highLevelClient = new RestHighLevelClient(lowLevelClient);

    LOGGER.info("Initialized connections to ElasticSearch: %s (port %d)", String.join(",", contactPoints), port);
  }
}

/**
 * 描述: Elasticsearch服务初始化
 * 时间: 2017年11月14日 上午10:55:02
 * @author yi.zhang
 */
public void init(){
    try {
        super.init();
        xclient = new RestHighLevelClient(super.getClient());
    } catch (Exception e) {
        logger.error("-----Elasticsearch Config init Error-----", e);
    }
}

/**
 * 描述: Elasticsearch服务初始化
 * 时间: 2017年11月14日 上午10:55:02
 * @author yi.zhang
 */
public void init(){
    try {
        super.init();
        xclient = new RestHighLevelClient(super.getClient());
    } catch (Exception e) {
        logger.error("-----Elasticsearch Config init Error-----", e);
    }
}

public void run(Configuration configuration, Environment environment) throws Exception {
  final CollectorRegistry collectorRegistry = new CollectorRegistry();
  collectorRegistry.register(new DropwizardExports(environment.metrics()));
  environment.admin()
      .addServlet("metrics", new MetricsServlet(collectorRegistry))
      .addMapping("/metrics");

  final PrometheusMetricsReporter reporter = PrometheusMetricsReporter.newMetricsReporter()
      .withCollectorRegistry(collectorRegistry)
      .withConstLabel("service", getName())
      .build();

  final Tracer tracer = getTracer();
  final Tracer metricsTracer = io.opentracing.contrib.metrics.Metrics.decorate(tracer, reporter);
  GlobalTracer.register(metricsTracer);

  final DynamicFeature tracing = new ServerTracingDynamicFeature.Builder(metricsTracer).build();
  environment.jersey().register(tracing);

  final HttpHost httpHost = new HttpHost("tweets-elasticsearch", 9200);
  final RestClientBuilder restClientBuilder =
      RestClient.builder(httpHost).setHttpClientConfigCallback(new TracingHttpClientConfigCallback(metricsTracer));
  final RestClient restClient = restClientBuilder.build();
  final RestHighLevelClient restHighLevelClient = new RestHighLevelClient(restClient);
  final TweetRepository tweetRepository = new ElasticsearchTweetRepository(restHighLevelClient);
  final TweetsResource tweetsResource = new TweetsResource(tweetRepository);
  environment.jersey().register(tweetsResource);
}

@Override
protected Void call() throws Exception {
    int performed = 0;

    try(RestHighLevelClient client = new RestHighLevelClient(RestClient.builder(new HttpHost(hostName, port, "http"))
                                                                        .setRequestConfigCallback(b -> b.setConnectTimeout(timeout).setSocketTimeout(timeout))
                                                                        .setMaxRetryTimeoutMillis(timeout));
        BulkProcessor processor = (new BulkProcessor.Builder(client::bulkAsync, new BulkProcessorListener(), new ThreadPool(Settings.builder()
                                                                                                                                     .put(Node.NODE_NAME_SETTING.getKey(), "high-level-client").build())))
                                     .setBulkActions(bulkCount)
                                     .build()){

        for(LogData log : logs){
            String index = "jvmul";
            if(log.getTime() != null){
                index += "-" + DateTimeFormatter.ISO_LOCAL_DATE.format(log.getTime());
            }
            else if(log.getUtcTime() != null){
                index += "-" + DateTimeFormatter.ISO_LOCAL_DATE.format(log.getUtcTime());
            }

            processor.add(new IndexRequest(index, "jvmul").source(buildContent(log)));
            updateProgress(++performed, logs.size());
            updateMessage(Integer.toString(performed) + " / " + Integer.toString(logs.size()));
        }

    }

    return null;
}

/**
 * This test documents docs/java-rest/high-level/document/delete.asciidoc
 */
public void testDelete() throws IOException {
    RestHighLevelClient client = highLevelClient();

    // tag::delete-request[]
    DeleteRequest request = new DeleteRequest(
        "index",    // <1>
        "type",     // <2>
        "id");      // <3>
    // end::delete-request[]

    // tag::delete-request-props[]
    request.timeout(TimeValue.timeValueSeconds(1));                     // <1>
    request.timeout("1s");                                              // <2>
    request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);    // <3>
    request.setRefreshPolicy("wait_for");                               // <4>
    request.version(2);                                                 // <5>
    request.versionType(VersionType.EXTERNAL);                          // <6>
    // end::delete-request-props[]

    // tag::delete-execute[]
    DeleteResponse response = client.delete(request);
    // end::delete-execute[]

    try {
        // tag::delete-notfound[]
        if (response.getResult().equals(DocWriteResponse.Result.NOT_FOUND)) {
            throw new Exception("Can't find document to be removed"); // <1>
        }
        // end::delete-notfound[]
    } catch (Exception ignored) { }

    // tag::delete-execute-async[]
    client.deleteAsync(request, new ActionListener<DeleteResponse>() {
        @Override
        public void onResponse(DeleteResponse deleteResponse) {
            // <1>
        }

        @Override
        public void onFailure(Exception e) {
            // <2>
        }
    });
    // end::delete-execute-async[]

    // tag::delete-conflict[]
    try {
        client.delete(request);
    } catch (ElasticsearchException exception) {
        if (exception.status().equals(RestStatus.CONFLICT)) {
            // <1>
        }
    }
    // end::delete-conflict[]

}

public RestHighLevelClient getHighLevelClient() {
  return highLevelClient;
}

public RestHighLevelClient getXClient(){
    return xclient;
}

public RestHighLevelClient getXClient(){
    return xclient;
}

public ElasticsearchTweetRepository(RestHighLevelClient restHighLevelClient) {
  this.restHighLevelClient = restHighLevelClient;
}