Use the Elasticsearch client RestHighLevelClient to connect to LindormSearch - Lindorm

Le Java High Level REST Client est un client Elasticsearch qui fournit une API de haut niveau pour la gestion des index et des documents. LindormSearch étant compatible avec les versions 7.10 d'Elasticsearch et antérieures, vous pouvez utiliser ce client pour vous connecter à LindormSearch et exécuter des requêtes et des recherches complexes sans modifier le code existant de votre application.

Le Java High Level REST Client assure la compatibilité ascendante : par exemple, la version 6.7.0 peut communiquer avec des clusters Elasticsearch 6.7.0 ou ultérieurs. Utilisez la version 7.10.0 ou une version antérieure pour vous connecter à LindormSearch.

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

JDK 1.8 ou version ultérieure installé
Moteur de recherche LindormSearch activé. Consultez le Guide d'activation.
Adresse IP de votre client ajoutée à la liste d'autorisation de l'instance Lindorm. Consultez la section Configuration des listes d'autorisation.

Ajout des dépendances

Pour un projet Maven, ajoutez les dépendances suivantes à votre fichier pom.xml :

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.10.0</version>
</dependency>
<dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-core</artifactId>
    <version>2.20.0</version>
</dependency>
<dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-api</artifactId>
    <version>2.20.0</version>
</dependency>

Connexion à LindormSearch

Utilisez RestClient.builder() pour créer un objet RestHighLevelClient. Le client s'appuie sur BasicCredentialsProvider pour l'authentification.

// Set the Elasticsearch-compatible endpoint and port for LindormSearch.
String search_url = "ld-t4n5668xk31ui****-proxy-search-public.lindorm.rds.aliyuncs.com";
int search_port = 30070;

// Set the username and password. Retrieve them from the Lindorm console:
// navigate to Database Connections > Search Engine tab.
String username = "user";
String password = "test";

final CredentialsProvider credentials_provider = new BasicCredentialsProvider();
credentials_provider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));

RestHighLevelClient highClient = new RestHighLevelClient(
    RestClient.builder(new HttpHost(search_url, search_port, "http"))
        .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                return httpClientBuilder.setDefaultCredentialsProvider(credentials_provider);
            }
        })
);

Paramètres de connexion

Paramètre	Description
`search_url`	Endpoint compatible Elasticsearch pour le moteur de recherche LindormSearch. Pour obtenir l'endpoint, consultez la section Adresse compatible Elasticsearch. Utilisez l'adresse VPC lors d'une connexion depuis un VPC, ou l'adresse Internet lors d'une connexion via le réseau public.
`search_port`	Port pour la compatibilité Elasticsearch sur LindormSearch : `30070`.
`username`	Nom d'utilisateur pour le moteur de recherche.
`password`	Mot de passe pour le moteur de recherche.

Choix du type de connexion réseau

VPC (recommandé) : Si votre application s'exécute sur une instance Elastic Compute Service (ECS) dans le même VPC que l'instance Lindorm, privilégiez la connexion via VPC pour une latence réduite et une sécurité accrue. Définissez search_url sur l'adresse VPC de l'endpoint compatible Elasticsearch.
Réseau public : Si votre application s'exécute en dehors d'Alibaba Cloud, activez d'abord l'endpoint public. Dans la console Lindorm, accédez à Database Connections, cliquez sur l'onglet Search Engine, puis cliquez sur Enable Public Endpoint dans le coin supérieur droit. Définissez ensuite search_url sur l'adresse Internet de l'endpoint compatible Elasticsearch.

Création d'un index

Utilisez CreateIndexRequest pour créer un index. L'exemple ci-dessous crée un index nommé lindorm_index avec 4 shards.

String index_name = "lindorm_index";

// Create a CreateIndexRequest and configure index settings.
CreateIndexRequest createIndexRequest = new CreateIndexRequest(index_name);
Map<String, Object> settingsMap = new HashMap<>();
settingsMap.put("index.number_of_shards", 4);
createIndexRequest.settings(settingsMap);

CreateIndexResponse createIndexResponse = highClient.indices().create(createIndexRequest, COMMON_OPTIONS);
if (createIndexResponse.isAcknowledged()) {
    System.out.println("Create index [" + index_name + "] successfully.");
}

Indexation d'un document

Utilisez IndexRequest pour écrire un seul document. Spécifiez un ID de document ou laissez-le vide pour que le système en génère un automatiquement (l'omission de l'ID peut améliorer les performances d'écriture).

// Specify the document ID.
String doc_id = "test";

// Build the document fields. Replace with actual field names and values for your use case.
Map<String, Object> jsonMap = new HashMap<>();
jsonMap.put("field1", "value1");
jsonMap.put("field2", "value2");

IndexRequest indexRequest = new IndexRequest(index_name);
indexRequest.id(doc_id).source(jsonMap);

IndexResponse indexResponse = highClient.index(indexRequest, COMMON_OPTIONS);
System.out.println("Index document with id[" + indexResponse.getId() + "] successfully.");

Indexation groupée de documents

Utilisez BulkProcessor avec bulkAsync() pour écrire efficacement un grand nombre de documents. BulkProcessor regroupe les requêtes d'indexation individuelles et les envoie par lot dès qu'un seuil configuré est atteint.

Configuration de BulkProcessor

Paramètre	Description	Exemple de valeur
`setConcurrentRequests`	Nombre maximal de requêtes groupées simultanées. Augmentez cette valeur pour améliorer le débit sous de fortes charges d'écriture. Valeur par défaut : `1`.	`10`
`setFlushInterval`	Intervalle de temps après lequel une requête groupée est envoyée, indépendamment de la taille ou du nombre. Utilisez ce paramètre comme filet de sécurité pour éviter que les données ne restent trop longtemps en mémoire tampon.	`5` secondes
`setBulkActions`	Nombre d'opérations individuelles déclenchant un vidage. Ajustez cette valeur en fonction de la taille moyenne de vos documents.	`5000`
`setBulkSize`	Taille totale des opérations mises en mémoire tampon déclenchant un vidage.	`5` Mo

int bulkTotal = 100000;
AtomicLong failedBulkItemCount = new AtomicLong();

BulkProcessor.Builder builder = BulkProcessor.builder(
    (request, bulkListener) -> highClient.bulkAsync(request, COMMON_OPTIONS, bulkListener),
    new BulkProcessor.Listener() {
        @Override
        public void beforeBulk(long executionId, BulkRequest request) {}

        @Override
        public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
            // Count failed items in the bulk response.
            for (BulkItemResponse bulkItemResponse : response) {
                if (bulkItemResponse.isFailed()) {
                    failedBulkItemCount.incrementAndGet();
                }
            }
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
            // If this callback fires, all requests in the bulk were not executed.
            if (null != failure) {
                failedBulkItemCount.addAndGet(request.numberOfActions());
            }
        }
    });

// Maximum concurrent bulk requests. Default is 1; increase for higher throughput.
builder.setConcurrentRequests(10);
// Flush thresholds — a bulk request is sent when any of these is met.
builder.setFlushInterval(TimeValue.timeValueSeconds(5));  // every 5 seconds
builder.setBulkActions(5000);                             // every 5,000 operations
builder.setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB)); // every 5 MB

BulkProcessor bulkProcessor = builder.build();
Random random = new Random();
for (int i = 0; i < bulkTotal; i++) {
    // Replace with actual field names and values for your use case.
    Map<String, Object> map = new HashMap<>();
    map.put("field1", random.nextInt() + "");
    map.put("field2", random.nextInt() + "");
    IndexRequest bulkItemRequest = new IndexRequest(index_name);
    bulkItemRequest.source(map);
    bulkProcessor.add(bulkItemRequest);
}

// Wait up to 120 seconds for all pending operations to complete.
bulkProcessor.awaitClose(120, TimeUnit.SECONDS);
long failure = failedBulkItemCount.get(),
     success = bulkTotal - failure;
System.out.println("Bulk using BulkProcessor finished with [" + success + "] requests succeeded, [" + failure + "] requests failed.");

Recherche de documents

Envoyez d'abord une requête d'actualisation pour rendre les données récemment écrites visibles, puis exécutez vos requêtes.

Par défaut, une requête de recherche renvoie au maximum 10 000 documents. Pour obtenir le nombre total exact lorsque plus de 10 000 documents correspondent, appelez searchSourceBuilder.trackTotalHits(true) avant d'exécuter la requête.

// Refresh the index to make written data searchable.
RefreshRequest refreshRequest = new RefreshRequest(index_name);
highClient.indices().refresh(refreshRequest, COMMON_OPTIONS);
System.out.println("Refresh on index [" + index_name + "] successfully.");

// Query all documents.
SearchRequest searchRequest = new SearchRequest(index_name);
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
// Uncomment the next line to get the exact total count when results exceed 10,000.
// searchSourceBuilder.trackTotalHits(true);
QueryBuilder queryMatchAllBuilder = new MatchAllQueryBuilder();
searchSourceBuilder.query(queryMatchAllBuilder);
searchRequest.source(searchSourceBuilder);
SearchResponse searchResponse = highClient.search(searchRequest, COMMON_OPTIONS);
long totalHit = searchResponse.getHits().getTotalHits().value;
System.out.println("Search query match all hits [" + totalHit + "] in total.");

// Query documents by ID.
QueryBuilder queryByIdBuilder = new MatchQueryBuilder("_id", doc_id);
searchSourceBuilder.query(queryByIdBuilder);
searchRequest.source(searchSourceBuilder);
searchResponse = highClient.search(searchRequest, COMMON_OPTIONS);
for (SearchHit searchHit : searchResponse.getHits()) {
    System.out.println("Search query by id response [" + searchHit.getSourceAsString() + "]");
}

Suppression de documents et d'index

Utilisez DeleteRequest pour supprimer un seul document et DeleteIndexRequest pour supprimer un index.

// Delete a single document by ID.
DeleteRequest deleteRequest = new DeleteRequest(index_name);
deleteRequest.id(doc_id);
DeleteResponse deleteResponse = highClient.delete(deleteRequest, COMMON_OPTIONS);
System.out.println("Delete document with id [" + deleteResponse.getId() + "] successfully.");

// Delete the index.
DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index_name);
AcknowledgedResponse deleteIndexResponse = highClient.indices().delete(deleteIndexRequest, COMMON_OPTIONS);
if (deleteIndexResponse.isAcknowledged()) {
    System.out.println("Delete index [" + index_name + "] successfully.");
}

highClient.close();

Exemple complet

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.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.MatchAllQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class RestHClientTest {
    private static final RequestOptions COMMON_OPTIONS;
    static {
        // Set the maximum response buffer size to 30 MB. The default is 100 MB.
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        builder.setHttpAsyncResponseConsumerFactory(
            new HttpAsyncResponseConsumerFactory
                .HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
        COMMON_OPTIONS = builder.build();
    }

    public static void main(String[] args) {
        // Set the Elasticsearch-compatible endpoint and port for LindormSearch.
        String search_url = "ld-t4n5668xk31ui****-proxy-search-public.lindorm.rds.aliyuncs.com";
        int search_port = 30070;

        // Set the username and password. Retrieve them from the Lindorm console.
        String username = "user";
        String password = "test";

        final CredentialsProvider credentials_provider = new BasicCredentialsProvider();
        credentials_provider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        RestHighLevelClient highClient = new RestHighLevelClient(
            RestClient.builder(new HttpHost(search_url, search_port, "http"))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setDefaultCredentialsProvider(credentials_provider);
                    }
                })
        );

        try {
            String index_name = "lindorm_index";

            // Create an index.
            CreateIndexRequest createIndexRequest = new CreateIndexRequest(index_name);
            Map<String, Object> settingsMap = new HashMap<>();
            settingsMap.put("index.number_of_shards", 4);
            createIndexRequest.settings(settingsMap);
            CreateIndexResponse createIndexResponse = highClient.indices().create(createIndexRequest, COMMON_OPTIONS);
            if (createIndexResponse.isAcknowledged()) {
                System.out.println("Create index [" + index_name + "] successfully.");
            }

            // Index a single document.
            // Specify the document ID. If you do not specify the document ID, an ID is automatically generated,
            // which can improve write performance.
            String doc_id = "test";
            Map<String, Object> jsonMap = new HashMap<>();
            jsonMap.put("field1", "value1");
            jsonMap.put("field2", "value2");
            IndexRequest indexRequest = new IndexRequest(index_name);
            indexRequest.id(doc_id).source(jsonMap);
            IndexResponse indexResponse = highClient.index(indexRequest, COMMON_OPTIONS);
            System.out.println("Index document with id[" + indexResponse.getId() + "] successfully.");

            // Bulk index documents using BulkProcessor.
            int bulkTotal = 100000;
            AtomicLong failedBulkItemCount = new AtomicLong();
            BulkProcessor.Builder builder = BulkProcessor.builder(
                (request, bulkListener) -> highClient.bulkAsync(request, COMMON_OPTIONS, bulkListener),
                new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId, BulkRequest request) {}

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                        for (BulkItemResponse bulkItemResponse : response) {
                            if (bulkItemResponse.isFailed()) {
                                failedBulkItemCount.incrementAndGet();
                            }
                        }
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                        if (null != failure) {
                            failedBulkItemCount.addAndGet(request.numberOfActions());
                        }
                    }
                });
            builder.setConcurrentRequests(10);
            builder.setFlushInterval(TimeValue.timeValueSeconds(5));
            builder.setBulkActions(5000);
            builder.setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB));
            BulkProcessor bulkProcessor = builder.build();
            Random random = new Random();
            for (int i = 0; i < bulkTotal; i++) {
                Map<String, Object> map = new HashMap<>();
                map.put("field1", random.nextInt() + "");
                map.put("field2", random.nextInt() + "");
                IndexRequest bulkItemRequest = new IndexRequest(index_name);
                bulkItemRequest.source(map);
                bulkProcessor.add(bulkItemRequest);
            }
            bulkProcessor.awaitClose(120, TimeUnit.SECONDS);
            long failure = failedBulkItemCount.get(),
                 success = bulkTotal - failure;
            System.out.println("Bulk using BulkProcessor finished with [" + success + "] requests succeeded, [" + failure + "] requests failed.");

            // Refresh the index to make written data searchable.
            RefreshRequest refreshRequest = new RefreshRequest(index_name);
            RefreshResponse refreshResponse = highClient.indices().refresh(refreshRequest, COMMON_OPTIONS);
            System.out.println("Refresh on index [" + index_name + "] successfully.");

            // Query all documents. By default, at most 10,000 results are returned.
            // To get the exact total count, call searchSourceBuilder.trackTotalHits(true).
            SearchRequest searchRequest = new SearchRequest(index_name);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            QueryBuilder queryMatchAllBuilder = new MatchAllQueryBuilder();
            searchSourceBuilder.query(queryMatchAllBuilder);
            searchRequest.source(searchSourceBuilder);
            SearchResponse searchResponse = highClient.search(searchRequest, COMMON_OPTIONS);
            long totalHit = searchResponse.getHits().getTotalHits().value;
            System.out.println("Search query match all hits [" + totalHit + "] in total.");

            // Query documents by ID.
            QueryBuilder queryByIdBuilder = new MatchQueryBuilder("_id", doc_id);
            searchSourceBuilder.query(queryByIdBuilder);
            searchRequest.source(searchSourceBuilder);
            searchResponse = highClient.search(searchRequest, COMMON_OPTIONS);
            for (SearchHit searchHit : searchResponse.getHits()) {
                System.out.println("Search query by id response [" + searchHit.getSourceAsString() + "]");
            }

            // Delete a document by ID.
            DeleteRequest deleteRequest = new DeleteRequest(index_name);
            deleteRequest.id(doc_id);
            DeleteResponse deleteResponse = highClient.delete(deleteRequest, COMMON_OPTIONS);
            System.out.println("Delete document with id [" + deleteResponse.getId() + "] successfully.");

            // Delete the index.
            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index_name);
            AcknowledgedResponse deleteIndexResponse = highClient.indices().delete(deleteIndexRequest, COMMON_OPTIONS);
            if (deleteIndexResponse.isAcknowledged()) {
                System.out.println("Delete index [" + index_name + "] successfully.");
            }

            highClient.close();
        } catch (Exception exception) {
            System.out.println("msg " + exception);
        }
    }
}

Le résultat attendu est le suivant :

Create index [lindorm_index] successfully.
Index document with id[test] successfully.
Bulk using BulkProcessor finished with [100000] requests succeeded, [0] requests failed.
Refresh on index [lindorm_index] successfully.
Search query match all hits [10000] in total.
Search query by id response [{"field1":"value1","field2":"value2"}]
Delete document with id [test] successfully.
Delete index [lindorm_index] successfully.

La requête match-all renvoie 10 000 résultats, même si 100 000 documents ont été écrits. Il s'agit de la limite de résultats par défaut. Pour obtenir le nombre total exact, appelez searchSourceBuilder.trackTotalHits(true) avant d'exécuter la requête.