在reactor中实现while循环以获取最新的elasticsearch索引

oymdgrw7 于 2021-06-15 发布在 ElasticSearch

关注(0)|答案(2)|浏览(520)

我在被动ElasticSearch中的索引名如下：

logs-2020.08.18
logs-2020.08.17
logs-2020.08.16

它将每天创建。
我想获取最新的索引名并使用reactiveelasticsearchclient或spring数据获取日志。有可能吗？
我在SpringWebFlux应用程序中尝试了以下方法：
我有以下代码段来查找索引可用性：

public Flux<Log> getLogFromLatestIndex(String serialId) {
    Calendar cal = Calendar.getInstance();
    String currentIndex = StringUtils.EMPTY;
    boolean indexExists = false;
    while (!indexExists) {
        currentIndex = String.format("logs-%s”, format(cal.getTime(), "yyyy.MM.dd"));
        indexExists = isIndexExists(currentIndex).block();
        cal.add(Calendar.DATE, -1); // Decrease day 1 until you find index
    }

    SearchQuery searchQuery = new NativeSearchQueryBuilder()
            .withQuery(matchQuery("serialId", serialId))
            .withIndices(currentIndex)
            .build();

    return reactiveElasticsearchTemplate.find(searchQuery, Log.class);
}

public Mono<Boolean> isIndexExists(String indexName) {
    return reactiveElasticsearchClient.indices().existsIndex(new GetIndexRequest().indices(indexName));
}

如何在不使用block的情况下获取布尔值

indexExists = isIndexExists(currentIndex).block();

显然，我会得到以下错误：

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-2

Java elasticsearch spring-webflux project-reactor reactive

来源：https://stackoverflow.com/questions/63503519/implement-while-loop-in-reactor-to-fetch-latest-elasticsearch-index

2条答案

按热度按时间

yk9xbfzb1#

应用程序.yml

spring.data.elasticsearch.client.reactive.endpoints: ip1:9200,ip2:9200,ip3:9200
spring.data.elasticsearch.rest.uris: ip1:9200,ip2:9200,ip3:9200

配置.java

@Configuration
public class ElasticSearchConfig extends AbstractReactiveElasticsearchConfiguration {

    @Value("${spring.data.elasticsearch.client.reactive.endpoints}")
    private String elasticSearchEndpoint;

    @Override
    public ReactiveElasticsearchClient reactiveElasticsearchClient() {
        final ClientConfiguration clientConfiguration = ClientConfiguration.builder()
            .connectedTo(elasticSearchEndpoint.split(","))
            .withWebClientConfigurer(webClient -> {
                ExchangeStrategies exchangeStrategies = ExchangeStrategies.builder()
                        .codecs(configurer -> configurer.defaultCodecs()
                                .maxInMemorySize(-1))
                        .build();
                return webClient.mutate().exchangeStrategies(exchangeStrategies).build();
            })
            .build();
    return ReactiveRestClients.create(clientConfiguration);

    }
}

控制器.java

@GetMapping("/getLog/{serialId}")
public Flux<Log> getLog(@PathVariable String serialId) {
    return loggerService.getLog(serialId);
}

其余的都是你的代码。我只是在Map里面打印索引名。虽然我在ElasticSearch中有索引logs-2020.08.21，但它一直在打印logs-2020.08.20、logs-2020.08.19、logs-2020.08.18等索引，最终抛出错误。
注意：当我在application.yml中尝试使用单ip时，我得到了相同的错误。

public Flux<Log> getLog(String serialId) {
    return Flux.generate(LocalDate::now, this::generateNextDate)
           .map(day -> {
                System.out.println(String.format("logs-%s", day.format(DateTimeFormatter.ofPattern("yyyy.MM.dd"))));
                return String.format("logs-%s", day.format(DateTimeFormatter.ofPattern("yyyy.MM.dd")));
            })               
           .flatMap(this::isIndexExists)
           .skipUntil(Tuple2::getT2) // check index exists boolean and drop non-existing ones
           .next() // takes first existing
           .flatMapMany(tuple -> findLogs(tuple.getT1(), serialId));
}

private LocalDate generateNextDate(LocalDate currentDay, SynchronousSink<LocalDate> sink) {
    sink.next(currentDay);
    return currentDay.minusDays(1);
}

private Mono<Tuple2<String, Boolean>> isIndexExists(String indexName) {
    return reactiveElasticsearchClient.indices().existsIndex(new GetIndexRequest().indices(indexName))
        .map(exists -> Tuples.of(indexName, exists));
}

private Flux<Log> findLogs(String index, String serialId) {
    // your other ES query here
}

赞(0）回复(0）举报 2021-06-15

jhkqcmku2#

你可以用 Flux.generate 以及 (take/skip)(Until/While) 在React堆里做一个循环。
笔记：
已替换 Calendar 与 LocalDate 因为它是不可变的，更适合于函数式/React式编程。
这个 isIndexExists 方法返回 Tuple 有索引名的引用，但显然可以根据需要用更具描述性的类来替换

public Flux<Log> getLog(String serialId) {
    return Flux.generate(LocalDate::now, this::generateNextDate)
               .map(day -> String.format("logs-%s", day.format(DateTimeFormatter.ofPattern("yyyy.MM.dd"))))
               .concatMap(this::isIndexExists)
               .skipUntil(Tuple2::getT2) // check index exists boolean and drop non-existing ones
               .next() // takes first existing
               .flatMapMany(tuple -> findLogs(tuple.getT1(), serialId));
}

private LocalDate generateNextDate(LocalDate currentDay, SynchronousSink<LocalDate> sink) {
    sink.next(currentDay);
    return currentDay.minusDays(1);
}

private Mono<Tuple2<String, Boolean>> isIndexExists(String indexName) {
    return reactiveElasticsearchClient.indices().existsIndex(new GetIndexRequest().indices(indexName))
            .map(exists -> Tuples.of(indexName, exists));
}

private Flux<Log> findLogs(String index, String serialId) {
    // your other ES query here
}

赞(0）回复(0）举报 2021-06-15

我来回答

在reactor中实现while循环以获取最新的elasticsearch索引

2条答案

相关问题

热门标签

最新问答