[JAVA]es整合多数据源

配置文件

spring:
elasticsearch:
  rest:
    uris: ${ES_URL:http://***:9200}
    username: ${ES_USERNAME:***}
    password: ${ES_PASSWORD:***}
    
elastic:
  masterHost: ${ELASTIC_MASTER_HOST}
  masterPort: ${ELASTIC_MASTER_PORT}
  masterUsername: ${ELASTIC_MASTER_USERNAME}
  masterPassword: ${ELASTIC_MASTER_PASSWORD}
  slaveHost: ${ELASTIC_SLAVE_HOST}
  slavePort: ${ELASTIC_SLAVE_PORT}
  slaveUsername: ${ELASTIC_SLAVE_USERNAME}
  slavePassword: ${ELASTIC_SLAVE_PASSWORD}

注入配置

@Data
@Configuration
@ConfigurationProperties(prefix = "elastic")
public class ElasticProperties {
    private String masterHost;
    private Integer masterPort;
    private String masterUsername;
    private String masterPassword;
    private String slaveHost;
    private Integer slavePort;
    private String slaveUsername;
    private String slavePassword;
}


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.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;

@Configuration
public class ElasticSearchConfig extends AbstractElasticsearchConfiguration {

    @Value("${spring.elasticsearch.rest.uris}")
    private String defaultUris;
    @Value("${spring.elasticsearch.rest.username}")
    private String defaultUsername;
    @Value("${spring.elasticsearch.rest.password}")
    private String defaultPassword;
    @Autowired
    private ElasticProperties elasticProperties;

    /**
     * 构建方式一
     */
    @Bean("defaultHighLevelClient")
    @Primary
    @Override
    public RestHighLevelClient elasticsearchClient() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(defaultUsername, defaultPassword));
        String[] arr = defaultUris.split(":");
        RestClientBuilder builder = RestClient.builder(new HttpHost(arr[1].replace("/", ""), Integer.parseInt(arr[2]), arr[0]))
                .setRequestConfigCallback(requestConfigBuilder -> {
                    requestConfigBuilder.setConnectTimeout(-1);
                    requestConfigBuilder.setSocketTimeout(-1);
                    requestConfigBuilder.setConnectionRequestTimeout(-1);
                    return requestConfigBuilder;
                })
                .setHttpClientConfigCallback(httpClientBuilder -> {
                    httpClientBuilder.disableAuthCaching();
                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                });

        return new RestHighLevelClient(builder);
    }

    /**
     * 构建方式二
     */
    @Bean("masterRestHighLevelClient")
    public RestHighLevelClient masterRestHighLevelClient() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(elasticProperties.getMasterUsername(), elasticProperties.getMasterPassword()));

        RestClientBuilder builder = RestClient.builder(new HttpHost(elasticProperties.getMasterHost(), elasticProperties.getMasterPort(), "http"))
                .setRequestConfigCallback(requestConfigBuilder -> {
                    requestConfigBuilder.setConnectTimeout(-1);
                    requestConfigBuilder.setSocketTimeout(-1);
                    requestConfigBuilder.setConnectionRequestTimeout(-1);
                    return requestConfigBuilder;
                })
                .setHttpClientConfigCallback(httpClientBuilder -> {
                    httpClientBuilder.disableAuthCaching();
                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                });

        return new RestHighLevelClient(builder);
    }

    @Bean("masterElasticsearchTemplate")
    public ElasticsearchRestTemplate masterElasticsearchTemplate() {
        return new ElasticsearchRestTemplate(masterRestHighLevelClient());
    }

    /**
     * 构建方式三
     */
    @Bean("slaveRestHighLevelClient")
    public RestHighLevelClient slaveRestHighLevelClient() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(elasticProperties.getSlaveUsername(), elasticProperties.getSlavePassword()));

        RestClientBuilder builder = RestClient.builder(new HttpHost(elasticProperties.getSlaveHost(), elasticProperties.getSlavePort(), "http"))
                .setRequestConfigCallback(requestConfigBuilder -> {
                    requestConfigBuilder.setConnectTimeout(-1);
                    requestConfigBuilder.setSocketTimeout(-1);
                    requestConfigBuilder.setConnectionRequestTimeout(-1);
                    return requestConfigBuilder;
                })
                .setHttpClientConfigCallback(httpClientBuilder -> {
                    httpClientBuilder.disableAuthCaching();
                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                });

        return new RestHighLevelClient(builder);
    }

    @Bean("slaveElasticsearchTemplate")
    public ElasticsearchRestTemplate slaveElasticsearchTemplate() {
        return new ElasticsearchRestTemplate(slaveRestHighLevelClient());
    }

}

注入使用

import cn.hutool.core.date.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.Query;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;

@Slf4j
@Component
public class ElasticOperator {

    @Resource(name = "masterElasticsearchTemplate")
    private ElasticsearchRestTemplate masterRestTemplate;

    @Resource(name = "slaveElasticsearchTemplate")
    private ElasticsearchRestTemplate slaveRestTemplate;

    private ElasticsearchRestTemplate currentTemplate;

    private ElasticsearchRestTemplate otherTemplate;

    private String CURRENT = "master";

    private String MASTER = "master";

    private String SLAVE = "slave";

    @PostConstruct
    private void init() {
        currentTemplate = masterRestTemplate;
        otherTemplate = slaveRestTemplate;
    }

    public void changeConnection(String symbol) {
        Date date = DateUtil.date();
        if (MASTER.equals(symbol)) {
            currentTemplate = masterRestTemplate;
            otherTemplate = slaveRestTemplate;
            CURRENT = "master";
            log.info("切换查询源成功，当前的主查询链接为：{}，切换时间为：{}", CURRENT, DateUtil.format(date, "yyyy-MM-dd HH:mm:ss"));
        } else if (SLAVE.equals(symbol)) {
            currentTemplate = slaveRestTemplate;
            otherTemplate = masterRestTemplate;
            CURRENT = "slave";
            log.info("切换查询源成功，当前的主查询链接为：{}，切换时间为：{}", CURRENT, DateUtil.format(date, "yyyy-MM-dd HH:mm:ss"));
        } else {
            log.error("切换查询源失败，传递的标识错误，当前的主查询链接为：{}，切换时间为：{}", CURRENT, DateUtil.format(date, "yyyy-MM-dd HH:mm:ss"));
        }

    }

    public <T> SearchHits<T> search(Query query, Class<T> clazz) {
        Long begin = System.currentTimeMillis();
        try {
            SearchHits<T> searchHits = currentTemplate.search(query, clazz);
            log.info("查询master节点es耗时：{}ms", (System.currentTimeMillis() - begin));
            return searchHits;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("查询主集群失败，查询备用集群，报错信息：{}", e.getMessage());
            SearchHits<T> searchHits = otherTemplate.search(query, clazz);
            log.info("查询slave节点es耗时：{}ms", (System.currentTimeMillis() - begin));
            return searchHits;
        }
    }
}