使用Quarkus在Elasticsearch进行响应式方法,案例分享！干货！

2021-09-09 16:55:30 浏览数 (4)

我已经实现了一项服务，Quarkus作为主要框架，Elasticsearch作为数据存储。在实现过程中，我萌生了写一篇关于如何使用Elasticsearch 的 Java High Level REST Client以反应式方式绑定 Quarkus 的想法。

开始对文章做笔记，将常用库（otaibe-commons-quarkus-elasticsearch模块）中常用的Elasticsearch相关代码分离出来存放在Github中。然后，我花了几个小时以 Quarkus 指南页面中的方式组装了一个简单的示例项目（也在 Github 中）  。目前，那里缺少 Elasticsearch 指南。

让我们继续更详细地解释如何连接 Quarkus 和 Elasticsearch。

创建 Quarkus 项目

mvn io.quarkus:quarkus-maven-plugin:1.0.1.Final:create \
    -DprojectGroupId=org.otaibe.quarkus.elasticsearch.example \
    -DprojectArtifactId=otaibe-quarkus-elasticsearch-example \
    -DclassName="org.otaibe.quarkus.elasticsearch.example.web.controller.FruitResource" \
    -Dpath="/fruits" \
    -Dextensions="resteasy-jackson,elasticsearch-rest-client"

Maven 设置

如您所见，Quarkus 中存在一个elasticsearch -rest-client  ；然而，这是一个 Elasticsearch Java 低级 REST 客户端。如果我们想使用 Elasticsearch Java High Level REST Client，我们只需要将它作为依赖添加到pom.xml文件中：

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
   <version>7.4.0</version>
</dependency>

请确保 Elasticsearch Java Low Level REST Client 的版本与 Elasticsearch Java High Level REST Client匹配。

由于我们以响应式方式使用 Elasticsearch，因此我更喜欢使用 Project Reactor。我们必须在依赖管理部分添加 BOM：

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-bom</artifactId>
    <version>Dysprosium-SR2</version>
    <type>pom</type>
    <scope>import</scope>
</dependency>

我们还必须添加 reactor-core 作为依赖项： 

<dependency>
    <groupId>io.projectreactor</groupId>
  <artifactId>reactor-core</artifactId>
</dependency>

我已经在一个库中分离了公共代码，所以我们应该将这个库添加到我们的示例项目中。为此，我们将使用Jitpack。这是一项很棒的服务。你只需要为 你的 Github 项目指出 正确的方法，它就会为它构建一个工件。这是我使用它的方式： 

<dependency>
    <groupId>com.github.tpenakov.otaibe-commons-quarkus</groupId>
    <artifactId>otaibe-commons-quarkus-core</artifactId>
    <version>elasticsearch-example.02</version>
</dependency>
<dependency>
    <groupId>com.github.tpenakov.otaibe-commons-quarkus</groupId>
    <artifactId>otaibe-commons-quarkus-elasticsearch</artifactId>
    <version>elasticsearch-example.02</version>
</dependency>
<dependency>
    <groupId>com.github.tpenakov.otaibe-commons-quarkus</groupId>
    <artifactId>otaibe-commons-quarkus-rest</artifactId>
    <version>elasticsearch-example.02</version>
</dependency>

通过 Docker 启动 Elasticsearch

此外，我们应该启动 Elastisearch。最简单的方法是通过 Docker 运行它：

docker run -it --rm=true --name elasticsearch_quarkus_test \
    -p 11027:9200 -p 11028:9300 \
    -e "discovery.type=single-node" \
    docker.elastic.co/elasticsearch/elasticsearch:7.4.0

连接到 Elasticsearch

让我们从将我们的服务连接到 Elasticsearch 开始——示例项目中的实现很简单——因此它将侦听 Quarkus 启动和关闭事件并初始化或终止连接：

package org.otaibe.quarkus.elasticsearch.example.service;
import io.quarkus.runtime.ShutdownEvent;
import io.quarkus.runtime.StartupEvent;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.otaibe.commons.quarkus.elasticsearch.client.service.AbstractElasticsearchService;
import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.event.Observes;
@ApplicationScoped
@Getter
@Setter
@Slf4j
public class ElasticsearchService extends AbstractElasticsearchService {
    public void init(@Observes StartupEvent event) {
        log.info("init started");
        super.init();
        log.info("init completed");
    }

    public void shutdown(@Observes ShutdownEvent event) {
        log.info("shutdown started");
        super.shutdown();
        log.info("shutdown completed");
    }
}

连接到 Elasticsearch 的实际工作是在AbstractElasticsearchService 中完成的： 

public abstract class AbstractElasticsearchService {
    @ConfigProperty(name = "service.elastic-search.hosts")
    String[] hosts;
    @ConfigProperty(name = "service.elastic-search.num-threads", defaultValue = "10")
    Optional<Integer> numThreads;
    private RestHighLevelClient restClient;
    private Sniffer sniffer;
    @PostConstruct
    public void init() {
        log.info("init started");
        List<HttpHost> httpHosts = Arrays.stream(hosts)
                .map(s -> StringUtils.split(s, ':'))
                .map(strings -> new HttpHost(strings[0], Integer.valueOf(strings[1])))
                .collect(Collectors.toList());
        RestClientBuilder builder = RestClient.builder(httpHosts.toArray(new HttpHost[httpHosts.size()]));
        getNumThreads().ifPresent(integer ->
                builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setDefaultIOReactorConfig(
                        IOReactorConfig
                                .custom()
                                .setIoThreadCount(integer)
                                .build())
                ));
        restClient = new RestHighLevelClient(builder);
        sniffer = Sniffer.builder(getRestClient().getLowLevelClient()).build();
        log.info("init completed");
    }
}

如您所见，此处的连接是按照Elasticsearch 文档 中建议的方式完成的。我的实现取决于两个配置属性：

属性文件：

service.elastic-search.hosts=localhost:11027

这是从 Docker 启动后的 Elasticsearch 连接字符串。第二个可选属性是：属性文件

service.elastic-search.num-threads

这是客户端所需的线程数。

创建 POJO

现在，让我们创建域对象（Fruit）：

package org.otaibe.quarkus.elasticsearch.example.domain;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor(staticName = "of")
public class Fruit {
    public static final String ID = "id";
    public static final String EXT_REF_ID = "ext_ref_id";
    public static final String NAME = "name";
    public static final String DESCRIPTION = "description";
    public static final String VERSION = "version";
    @JsonProperty(ID)
    public String id;
    @JsonProperty(EXT_REF_ID)
    public String extRefId;
    @JsonProperty(NAME)
    public String name;
    @JsonProperty(DESCRIPTION)
    public String description;
    @JsonProperty(VERSION)
    public Long version;
}

创建和实现 DAO

创建索引

让我们创建 FruitDaoImpl。它是一个高级类，用于填充 AbstractElasticsearchReactiveDaoImplementation 并实现所需的业务逻辑。这里的另一个重要部分是为 Fruit 类创建索引：

@Override
protected Mono<Boolean> createIndex() {
    CreateIndexRequest request = new CreateIndexRequest(getTableName());
    Map<String, Object> mapping = new HashMap();
    Map<String, Object> propsMapping = new HashMap<>();
    propsMapping.put(Fruit.ID, getKeywordTextAnalizer());
    propsMapping.put(Fruit.EXT_REF_ID, getKeywordTextAnalizer());
    propsMapping.put(Fruit.NAME, getTextAnalizer(ENGLISH));
    propsMapping.put(Fruit.DESCRIPTION, getTextAnalizer(ENGLISH));
    propsMapping.put(Fruit.VERSION, getLongFieldType());
    mapping.put(PROPERTIES, propsMapping);
    request.mapping(mapping);
    return createIndex(request);
}

对 Elasticsearch 的真正创建索引调用是在父类 ( AbstractElasticsearchReactiveDaoImplementation ) 中实现的： 

protected Mono<Boolean> createIndex(CreateIndexRequest request) {
    return Flux.<Boolean>create(fluxSink -> getRestClient().indices().createAsync(request, RequestOptions.DEFAULT, new ActionListener<CreateIndexResponse>() {
        @Override
        public void onResponse(CreateIndexResponse createIndexResponse) {
            log.info("CreateIndexResponse: {}", createIndexResponse);
            fluxSink.next(createIndexResponse.isAcknowledged());
            fluxSink.complete();
        }
        @Override
        public void onFailure(Exception e) {
            log.error("unable to create index", e);
            fluxSink.error(new RuntimeException(e));
        }
    }))
            .next();
}

玩转 DAO

大多数 CRUD 操作在AbstractElasticsearchReactiveDaoImplementation中实现 。

它有  save、   update、  findById和  deleteById 公共方法。它也有findByExactMatch和 findByMatch保护方法。FindBy*当需要填充业务逻辑时，这些 方法在后代类中非常有用。

业务查找方法在FruitDaoImpl 类中实现 ：

public Flux<Fruit> findByExternalRefId(String value) {
    return findByMatch(Fruit.EXT_REF_ID, value);
}
public Flux<Fruit> findByName(String value) {
    return findByMatch(Fruit.NAME, value);
}
public Flux<Fruit> findByDescription(String value) {
    return findByMatch(Fruit.NAME, value);
}
public Flux<Fruit> findByNameOrDescription(String value) {
    Map<String, Object> query = new HashMap<>();
    query.put(Fruit.NAME, value);
    query.put(Fruit.DESCRIPTION, value);
    return findByMatch(query);
}

在Service类中封装 DAO

FruitDaoImpl 封装在 FruitService 中：

@ApplicationScoped
@Getter
@Setter
@Slf4j
public class FruitService {
    @Inject
    FruitDaoImpl dao;
    public Mono<Fruit> save(Fruit entity) {
        return getDao().save(entity);
    }
    public Mono<Fruit> findById(Fruit entity) {
        return getDao().findById(entity);
    }

    public Mono<Fruit> findById(String id) {
        return Mono.just(Fruit.of(id, null, null, null, null))

                .flatMap(entity -> findById(entity));
    }
    public Flux<Fruit> findByExternalRefId(String value) {
        return getDao().findByExternalRefId(value);
    }
    public Flux<Fruit> findByName(String value) {
        return getDao().findByName(value);
    }
    public Flux<Fruit> findByDescription(String value) {
        return getDao().findByDescription(value);
    }
    public Flux<Fruit> findByNameOrDescription(String value) {
        return getDao().findByNameOrDescription(value);
    }
    public Mono<Boolean> delete(Fruit entity) {
        return Mono.just(entity.getId())
                .filter(s -> StringUtils.isNotBlank(s))
                .flatMap(s -> getDao().deleteById(entity))
                .defaultIfEmpty(false);
    }
}

测试 FruitService

该 FruitServiceTests 写入，以测试基本功能。它还用于确保 Fruit 类字段被正确索引并且全文搜索按预期工作：

@Test
public void manageFruitTest() {
    Fruit apple = getTestUtils().createApple();
    Fruit apple1 = getFruitService().save(apple).block();
    Assertions.assertNotNull(apple1.getId());
    Assertions.assertTrue(apple1.getVersion() > 0);
    log.info("saved result: {}", getJsonUtils().toStringLazy(apple1));
    List<Fruit> fruitList = getFruitService().findByExternalRefId(TestUtils.EXT_REF_ID).collectList().block();
    Assertions.assertTrue(fruitList.size() > 0);
    List<Fruit> fruitList1 = getFruitService().findByNameOrDescription("bulgaria").collectList().block();
    Assertions.assertTrue(fruitList1.size() > 0);
    //Ensure that the full text search is working - it is 'Apples' in description
    List<Fruit> fruitList2 = getFruitService().findByDescription("apple").collectList().block();
    Assertions.assertTrue(fruitList2.size() > 0);
    //Ensure that the full text search is working - it is 'Apple' in name
    List<Fruit> fruitList3 = getFruitService().findByName("apples").collectList().block();
    Assertions.assertTrue(fruitList3.size() > 0);
    Boolean deleteAppleResult = getFruitService().getDao().deleteById(apple1).block();
    Assertions.assertTrue(deleteAppleResult);
}

添加 REST 端点

因为这是一个示例项目，完整的 CRUD 功能不会作为 REST 端点添加。只有save和 findById被添加为 REST 端点。它们被添加到 FruitResource 中。那里的方法返回 CompletionStage<Response>，这确保我们的应用程序中不会有阻塞的线程。

测试 REST 端点

 添加FruitResourceTest以测试 RESTendpoints：

package org.otaibe.quarkus.elasticsearch.example.web.controller;
import io.quarkus.test.junit.QuarkusTest;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.microprofile.config.inject.ConfigProperty;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.otaibe.commons.quarkus.core.utils.JsonUtils;
import org.otaibe.quarkus.elasticsearch.example.domain.Fruit;
import org.otaibe.quarkus.elasticsearch.example.service.FruitService;
import org.otaibe.quarkus.elasticsearch.example.utils.TestUtils;
import javax.inject.Inject;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriBuilder;
import java.net.URI;
import java.util.Optional;
import static io.restassured.RestAssured.given;
@QuarkusTest
@Getter(value = AccessLevel.PROTECTED)
@Slf4j
public class FruitResourceTest {
    @ConfigProperty(name = "service.http.host")
    Optional<URI> host;
    @Inject
    TestUtils testUtils;
    @Inject
    JsonUtils jsonUtils;
    @Inject
    FruitService service;
    @Test
    public void restEndpointsTest() {
        log.info("restEndpointsTest start");
        Fruit apple = getTestUtils().createApple();
        Fruit savedApple = given()
                .when()
                .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON)
                .body(apple)
                .post(getUri(FruitResource.ROOT_PATH))
                .then()
                .statusCode(200)
                .extract()
                .as(Fruit.class);
        String id = savedApple.getId();
        Assertions.assertTrue(StringUtils.isNotBlank(id));
        URI findByIdPath = UriBuilder.fromPath(FruitResource.ROOT_PATH)
                .path(id)
                .build();
        Fruit foundApple = given()
                .when().get(getUri(findByIdPath.getPath()).getPath())
                .then()
                .statusCode(200)
                .extract()
                .as(Fruit.class);
        Assertions.assertEquals(savedApple, foundApple);
        Boolean deleteResult = getService().delete(foundApple).block();
        Assertions.assertTrue(deleteResult);
        given()
                .when().get(findByIdPath.getPath())
                .then()
                .statusCode(Response.Status.NOT_FOUND.getStatusCode()) ;
        log.info("restEndpointsTest end");
    }
    private URI getUri(String path) {
        return getUriBuilder(path)
                .build();
    }
    private UriBuilder getUriBuilder(String path) {
        return getHost()
                .map(uri -> UriBuilder.fromUri(uri))
                .map(uriBuilder -> uriBuilder.path(path))
                .orElse(UriBuilder
                        .fromPath(path)
                );
    }
}

构建本地可执行文件

在构建本机可执行文件之前，我们必须注册我们的 Fruit 域对象。这样做的原因是我们的 FruitResource 返回 CompletionStage<Response>，因此，应用程序的实际返回类型是未知的，因此我们必须显式注册它以进行反射。在 Quarkus 中至少有两种方法可以做到这一点：

  1. 通过 @RegisterForReflection 注释。
  2. 通过 反射-config.json。

我个人更喜欢第二种方法，因为您要注册的类可能在第三方库中，并且不可能将 @RegisterForReflection 放在 那里。

现在，  reflection-config.json 看起来像这样：

[
  {
    "name" : "org.otaibe.quarkus.elasticsearch.example.domain.Fruit",
    "allDeclaredConstructors" : true,
    "allPublicConstructors" : true,
    "allDeclaredMethods" : true,
    "allPublicMethods" : true,
    "allDeclaredFields" : true,
    "allPublicFields" : true
  }
]

下一步是让 Quarkus 知道 reflection-config.json 文件。您应该将此行添加到pom.xml文件中的 native配置文件中： 

<quarkus.native.additional-build-args>-H:ReflectionConfigurationFiles=${project.basedir}/src/main/resources/reflection-config.json</quarkus.native.additional-build-args>

您现在可以构建您的本机应用程序： 

mvn clean package -Pnative

并启动它： 

./target/otaibe-quarkus-elasticsearch-example-1.0-SNAPSHOT-runner

该服务将在http://localhost:11025上可用，因为这是application.properties 中明确指定的端口。 

quarkus.http.port=11025

测试本机构建

该 FruitResourceTest 预计以下可选属性：

属性文件：

service.http.host

如果存在，测试请求将命中指定的主机。如果您启动本机可执行文件：

shell:

./target/otaibe-quarkus-elasticsearch-example-1.0-SNAPSHOT-runner

并使用以下代码执行测试/构建：

shell:

mvn package -D %test.service.http .host = http://localhost:11025

测试将针对本机构建运行。

结论

我惊喜地发现 Elasticsearch 与 Quarkus 一起开箱即用，可以编译为本地代码，结合通过Project Reactor 的反应式实现 ，将使应用程序的占用空间几乎微不足道。



Java 计算机原理 编程基础 大数据

1 人点赞