代码语言：javascript复制
package com.jinw.cms.config;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
import org.apache.http.HttpHeaders;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponseInterceptor;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.message.BasicHeader;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.ssl.SSLContexts;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder.HttpClientConfigCallback;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.util.StringUtils;
import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.InputStream;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.Certificate;
import java.security.cert.CertificateException;
import java.security.cert.CertificateFactory;
import java.util.Arrays;
@ConfigurationProperties(prefix = "spring.elasticsearch") //配置的前缀
@Configuration
@Slf4j
public class ElasticsearchConfig {
@Setter
private String uris;
@Setter
private String username;
@Setter
private String password;
/**
* 解析配置的字符串，转为HttpHost对象数组
*
* @return
*/
private HttpHost[] toHttpHost() {
if (!StringUtils.hasLength(uris)) {
throw new RuntimeException("invalid elasticsearch configuration");
}
String[] hostArray = uris.split(",");
HttpHost[] httpHosts = new HttpHost[hostArray.length];
HttpHost httpHost;
for (int i = 0; i < hostArray.length; i ) {
String[] strings = hostArray[i].split(":");
httpHost = new HttpHost(strings[0], Integer.parseInt(strings[1]), "http");
httpHosts[i] = httpHost;
}
return httpHosts;
}
@Bean("clientByPasswd")
public ElasticsearchClient clientByPasswd() throws Exception {
ElasticsearchTransport transport = getElasticsearchTransport(username, password, toHttpHost());
return new ElasticsearchClient(transport);
}
private static SSLContext buildSSLContext() {
ClassPathResource resource = new ClassPathResource("es01.crt");
SSLContext sslContext = null;
try {
CertificateFactory factory = CertificateFactory.getInstance("X.509");
Certificate trustedCa;
try (InputStream is = resource.getInputStream()) {
trustedCa = factory.generateCertificate(is);
}
KeyStore trustStore = KeyStore.getInstance("pkcs12");
trustStore.load(null, null);
trustStore.setCertificateEntry("ca", trustedCa);
SSLContextBuilder sslContextBuilder = SSLContexts.custom()
.loadTrustMaterial(trustStore, null);
sslContext = sslContextBuilder.build();
} catch (CertificateException | IOException | KeyStoreException | NoSuchAlgorithmException |
KeyManagementException e) {
log.error("ES连接认证失败", e);
}
return sslContext;
}
private static ElasticsearchTransport getElasticsearchTransport(String username, String passwd, HttpHost... hosts) {
// 账号密码的配置
final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, passwd));
// 自签证书的设置，并且还包含了账号密码
HttpClientConfigCallback callback = httpAsyncClientBuilder -> httpAsyncClientBuilder
// .setSSLContext(buildSSLContext()) // 证书式认证方式
.setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE)
.setDefaultCredentialsProvider(credentialsProvider)
// Todo 这里是关键 ######################解决 X-Elastic-Product 问题开始
.setDefaultHeaders(
Arrays.asList(
new BasicHeader(
HttpHeaders.CONTENT_TYPE, ContentType.APPLICATION_JSON.toString())))
.addInterceptorLast(
(HttpResponseInterceptor)
(response, context) ->
response.addHeader("X-Elastic-Product", "Elasticsearch"));
// X-Elastic-Product end
// 用builder创建RestClient对象
RestClient client = RestClient
.builder(hosts)
.setHttpClientConfigCallback(callback)
.build();
return new RestClientTransport(client, new JacksonJsonpMapper());
}
private static ElasticsearchTransport getElasticsearchTransport(String apiKey, HttpHost... hosts) {
// 将ApiKey放入header中
Header[] headers = new Header[]{new BasicHeader("Authorization", "ApiKey " apiKey)};
// es自签证书的设置
HttpClientConfigCallback callback = httpAsyncClientBuilder -> httpAsyncClientBuilder
.setSSLContext(buildSSLContext())
.setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
// 用builder创建RestClient对象
RestClient client = RestClient
.builder(hosts)
.setHttpClientConfigCallback(callback)
.setDefaultHeaders(headers)
.build();
return new RestClientTransport(client, new JacksonJsonpMapper());
}
}spring:
elasticsearch:
uris: 127.0.0.1:9200
username: elastic
password: 123456
connection-timeout: 10000
socket-timeout: 30spackage com.jinw.cms.service.impl;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.ElasticsearchException;
import co.elastic.clients.elasticsearch._types.mapping.Property;
import co.elastic.clients.elasticsearch.core.GetResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.jinw.cms.constants.*;
import com.jinw.cms.content.IContent;
import com.jinw.cms.content.type.IContentType;
import com.jinw.cms.entity.CmsArticle;
import com.jinw.cms.entity.CmsCategory;
import com.jinw.cms.entity.CmsSite;
import com.jinw.cms.entity.ESContent;
import com.jinw.cms.mapper.CmsCategoryMapper;
import com.jinw.cms.service.ICmsArticleService;
import com.jinw.cms.service.ICmsSiteService;
import com.jinw.utils.cms.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@Slf4j
@RequiredArgsConstructor
@Service
public class ContentIndexService implements CommandLineRunner {
private final ICmsSiteService siteService;
private final CmsCategoryMapper cmsCategoryMapper;
private final ICmsArticleService contentService;
private final ElasticsearchClient esClient;
private void createIndex() throws IOException {
// 创建索引
Map<String, Property> properties = new HashMap<>();
properties.put("catalogAncestors", Property.of(fn -> fn.keyword(b -> b
.ignoreAbove(500) // 指定字符串字段的最大长度。超过该长度的字符串将被截断或忽略。
)));
properties.put("contentType", Property.of(fn -> fn.keyword(b -> b
.ignoreAbove(20)
)));
properties.put("logo", Property.of(fn -> fn.keyword(b -> b
.ignoreAbove(256)
)));
properties.put("title", Property.of(fn -> fn.text(b -> b
.store(true) // 是否存储在索引中
.analyzer(SearchConsts.IKAnalyzeType_Smart)
)));
properties.put("fullText", Property.of(fn -> fn.text(b -> b
.analyzer(SearchConsts.IKAnalyzeType_Smart)
)));
CreateIndexResponse response = esClient.indices().create(fn -> fn
.index(ESContent.INDEX_NAME)
.mappings(mb -> mb.properties(properties)));
Assert.isTrue(response.acknowledged(), () -> new RuntimeException("Create Index[cms_article] failed."));
}
public void recreateIndex(CmsSite site) throws IOException {
boolean exists = esClient.indices().exists(fn -> fn.index(ESContent.INDEX_NAME)).value();
if (!exists) {
this.createIndex();
} else {
// 删除站点索引文档数据
long total = this.contentService.getContentMapper().selectCount(Wrappers.lambdaQuery(CmsArticle.class).eq(CmsArticle::getSiteId, site.getId()));
long pageSize = 1000;
for (int i = 0; i * pageSize < total; i ) {
List<String> contentIds = this.contentService.getContentMapper().selectPage(
new Page<>(i, pageSize, false), Wrappers.lambdaQuery(CmsArticle.class).eq(CmsArticle::getSiteId, site.getId())
).getRecords().stream().map(CmsArticle::getId).collect(Collectors.toList());
deleteContentDoc(contentIds);
}
}
}
/**
* 创建/更新内容索引Document
*/
public void createContentDoc(IContent<?> content) {
// 判断栏目/站点配置是否生成索引
String enableIndex = EnableIndexProperty.getValue(content.getCatalog().getConfigProps(),
content.getSite().getConfigProps());
if (YesOrNo.isNo(enableIndex)) {
return;
}
try {
esClient.update(fn -> fn
.index(ESContent.INDEX_NAME)
.id(content.getContentEntity().getId().toString())
.doc(newESContentDoc(content))
.docAsUpsert(true), ESContent.class);
} catch (ElasticsearchException | IOException e) {
// AsyncTaskManager.addErrMessage(e.getMessage());
e.printStackTrace();
}
}
private void batchContentDoc(CmsSite site, CmsCategory catalog, List<CmsArticle> contents) {
if (contents.isEmpty()) {
return;
}
List<BulkOperation> bulkOperationList = new ArrayList<>(contents.size());
for (CmsArticle xContent : contents) {
// 判断栏目/站点配置是否生成索引
String enableIndex = EnableIndexProperty.getValue(catalog.getConfigProps(), site.getConfigProps());
if (YesOrNo.isYes(enableIndex)) {
IContentType contentType = ContentCoreUtils.getContentType(xContent.getContentType());
IContent<?> icontent = contentType.loadContent(xContent);
BulkOperation bulkOperation = BulkOperation.of(b ->
b.update(up -> up.index(ESContent.INDEX_NAME)
.id(xContent.getId().toString())
.action(action -> action.docAsUpsert(true).doc(newESContentDoc(icontent))))
// b.create(co -> co.index(ESContent.INDEX_NAME)
// .id(xContent.getContentId().toString()).document(newESContent(icontent)))
);
bulkOperationList.add(bulkOperation);
}
}
if (bulkOperationList.isEmpty()) {
return;
}
// 批量新增索引
try {
esClient.bulk(bulk -> bulk.operations(bulkOperationList));
} catch (ElasticsearchException | IOException e) {
// AsyncTaskManager.addErrMessage(e.getMessage());
e.printStackTrace();
}
}
/**
* 删除内容索引
*/
public void deleteContentDoc(List<String> contentIds) throws ElasticsearchException, IOException {
List<BulkOperation> bulkOperationList = contentIds.stream().map(contentId -> BulkOperation
.of(b -> b.delete(dq -> dq.index(ESContent.INDEX_NAME).id(contentId.toString())))).collect(Collectors.toList());
this.esClient.bulk(bulk -> bulk.operations(bulkOperationList));
}
public void rebuildCatalog(CmsCategory catalog, boolean includeChild) {
CmsSite site = this.siteService.getSite(catalog.getSiteId());
String enableIndex = EnableIndexProperty.getValue(catalog.getConfigProps(), site.getConfigProps());
if (YesOrNo.isYes(enableIndex)) {
LambdaQueryWrapper<CmsArticle> q = new LambdaQueryWrapper<CmsArticle>()
.ne(CmsArticle::getCopyType, ContentCopyType.Mapping)
.eq(CmsArticle::getStatus, ContentStatus.PUBLISHED)
.eq(!includeChild, CmsArticle::getCategoryId, catalog.getId())
.likeRight(includeChild, CmsArticle::getCategoryAncestors, catalog.getAncestors());
long total = this.contentService.getContentMapper().selectCount(q);
long pageSize = 200;
for (int i = 0; i * pageSize < total; i ) {
Page<CmsArticle> page = contentService.getContentMapper().selectPage(new Page<>(i, pageSize, false), q);
batchContentDoc(site, catalog, page.getRecords());
}
}
}
/**
* 重建指定站点所有内容索引
*/
public void rebuildAll(CmsSite site) throws IOException {
// 先重建索引
recreateIndex(site);
List<CmsCategory> catalogs = cmsCategoryMapper.selectList(Wrappers.lambdaQuery());
for (CmsCategory category : catalogs) {
LambdaQueryWrapper<CmsArticle> q = new LambdaQueryWrapper<CmsArticle>()
.eq(CmsArticle::getSiteId, site.getId())
.ne(CmsArticle::getCopyType, ContentCopyType.Mapping)
.eq(CmsArticle::getStatus, ContentStatus.PUBLISHED)
.eq(CmsArticle::getCategoryId, category.getId());
long total = contentService.getContentMapper().selectCount(q);
int pageSize = 200;
int count = 1;
for (int i = 0; (long) i * pageSize < total; i ) {
log.debug((int) (count * 100 / total) "正在重建栏目【" category.getName() "】内容索引");
Page<CmsArticle> page = contentService.getContentMapper().selectPage(new Page<>(i, pageSize, false), q);
batchContentDoc(site, category, page.getRecords());
// AsyncTaskManager.checkInterrupt(); // 允许中断
}
}
log.debug("100% 重建全站索引完成");
}
/**
* 获取指定内容索引详情
*
* @param contentId 内容ID
* @return 索引Document详情
*/
public ESContent getContentDocDetail(String contentId) throws ElasticsearchException, IOException {
GetResponse<ESContent> res = this.esClient.get(qb -> qb.index(ESContent.INDEX_NAME).id(contentId.toString()),
ESContent.class);
return res.source();
}
private Map<String, Object> newESContentDoc(IContent<?> content) {
Map<String, Object> data = new HashMap<>();
data.put("contentId", content.getContentEntity().getId());
data.put("contentType", content.getContentEntity().getContentType());
data.put("siteId", content.getSiteId());
data.put("catalogId", content.getCatalogId());
data.put("catalogAncestors", content.getContentEntity().getCategoryAncestors());
data.put("author", content.getContentEntity().getAuthor());
data.put("editor", content.getContentEntity().getEditor());
data.put("keywords", StringUtils.join(content.getContentEntity().getKeywords()));
data.put("tags", StringUtils.join(content.getContentEntity().getTags()));
data.put("createTime", content.getContentEntity().getCreateTime());
data.put("logo", content.getContentEntity().getLogo());
data.put("status", content.getContentEntity().getStatus());
data.put("publishDate", content.getContentEntity().getPublishDate().toEpochSecond(ZoneOffset.UTC));
data.put("link", InternalUrlUtils.getInternalUrl(InternalDataType_Content.ID, content.getContentEntity().getId()));
data.put("title", content.getContentEntity().getTitle());
data.put("summary", content.getContentEntity().getSummary());
data.put("fullText", content.getFullText());
// 扩展模型数据
// this.extendModelService.getModelData(content.getContentEntity()).forEach(fd -> {
// data.put(fd.getFieldName(), fd.getValue());
// });
return data;
}
public boolean isElasticSearchAvailable() {
try {
return esClient.ping().value();
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
@Override
public void run(String... args) throws Exception {
if (isElasticSearchAvailable()) {
boolean exists = esClient.indices().exists(fn -> fn.index(ESContent.INDEX_NAME)).value();
if (!exists) {
this.createIndex(); // 创建内容索引库
}
} else {
log.warn("ES service not available!");
}
}
} <elasticsearch-java>8.6.2</elasticsearch-java> <dependency>
<groupId>co.elastic.clients</groupId>
<artifactId>elasticsearch-java</artifactId>
<version>${elasticsearch-java}</version>
<!--排除这些组件-->
<exclusions>
<exclusion>
<artifactId>jakarta.json-api</artifactId>
<groupId>jakarta.json</groupId>
</exclusion>
<exclusion>
<artifactId>elasticsearch-rest-client</artifactId>
<groupId>org.elasticsearch.client</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
<version>${elasticsearch-java}</version>
</dependency>
