Elasticsearch 搭建(亲测) - 实践

我来为您详细介绍Elasticsearch 7.17的下载、集群安装、客户端操作和Spring Boot集成。

1. Elasticsearch 7.17 下载和集群安装

1.1 下载Elasticsearch 7.17.10

bash

# 创建安装目录
mkdir -p /opt/elasticsearch
cd /opt/elasticsearch# 下载Elasticsearch 7.17.10
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.10-linux-x86_64.tar.gz# 解压
tar -xzf elasticsearch-7.17.10-linux-x86_64.tar.gz
cd elasticsearch-7.17.10# 创建数据和日志目录
mkdir -p /data/elasticsearch/{data,logs}

1.2 创建Elasticsearch系统用户

bash

# 创建elasticsearch用户和组
groupadd elasticsearch
useradd -g elasticsearch elasticsearch# 设置目录权限
chown -R elasticsearch:elasticsearch /opt/elasticsearch
chown -R elasticsearch:elasticsearch /data/elasticsearch

1.3 集群配置文件

节点1配置 (node-1)

bash

# 创建配置文件
cat > /opt/elasticsearch/elasticsearch-7.17.10/config/elasticsearch.yml << EOF
# 集群名称
cluster.name: my-es-cluster# 节点名称
node.name: node-1# 是否为主节点
node.master: true
node.data: true# 数据存储路径
path.data: /data/elasticsearch/data
path.logs: /data/elasticsearch/logs# 网络设置
network.host: 0.0.0.0
http.port: 9200
transport.port: 9300# 发现设置
discovery.seed_hosts: ["192.168.1.101:9300", "192.168.1.102:9300", "192.168.1.103:9300"]
cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]# 安全设置（生产环境建议开启）
xpack.security.enabled: false
xpack.monitoring.collection.enabled: true# 其他优化
bootstrap.memory_lock: true
thread_pool.write.queue_size: 1000
EOF

节点2配置 (node-2)

bash

cat > /opt/elasticsearch/elasticsearch-7.17.10/config/elasticsearch.yml << EOF
cluster.name: my-es-cluster
node.name: node-2
node.master: true
node.data: true
path.data: /data/elasticsearch/data
path.logs: /data/elasticsearch/logs
network.host: 0.0.0.0
http.port: 9201
transport.port: 9301
discovery.seed_hosts: ["192.168.1.101:9300", "192.168.1.102:9301", "192.168.1.103:9302"]
cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]
xpack.security.enabled: false
bootstrap.memory_lock: true
EOF

节点3配置 (node-3)

bash

cat > /opt/elasticsearch/elasticsearch-7.17.10/config/elasticsearch.yml << EOF
cluster.name: my-es-cluster
node.name: node-3
node.master: true
node.data: true
path.data: /data/elasticsearch/data
path.logs: /data/elasticsearch/logs
network.host: 0.0.0.0
http.port: 9202
transport.port: 9302
discovery.seed_hosts: ["192.168.1.101:9300", "192.168.1.102:9301", "192.168.1.103:9302"]
cluster.initial_master_nodes: ["node-1", "node-2", "node-3"]
xpack.security.enabled: false
bootstrap.memory_lock: true
EOF

1.4 JVM配置

bash

# 编辑JVM参数
cat > /opt/elasticsearch/elasticsearch-7.17.10/config/jvm.options << EOF
-Xms2g
-Xmx2g
-XX:+UseConcMarkSweepGC
-XX:CMSInitiatingOccupancyFraction=75
-XX:+UseCMSInitiatingOccupancyOnly
-XX:+AlwaysPreTouch
-Dfile.encoding=UTF-8
-Djava.awt.headless=true
EOF

1.5 系统参数优化

bash

# 增加文件描述符限制
echo 'elasticsearch soft nofile 65536' >> /etc/security/limits.conf
echo 'elasticsearch hard nofile 65536' >> /etc/security/limits.conf# 增加虚拟内存映射
echo 'vm.max_map_count=262144' >> /etc/sysctl.conf
sysctl -p# 禁用交换分区
swapoff -a

1.6 启动集群

bash

# 切换到elasticsearch用户
su - elasticsearch# 启动各个节点（在不同服务器上执行）
cd /opt/elasticsearch/elasticsearch-7.17.10
./bin/elasticsearch -d# 检查进程
ps aux | grep elasticsearch# 检查集群状态
curl -X GET "http://localhost:9200/_cluster/health?pretty"

2. 客户端命令操作

2.1 集群状态监控命令

bash

# 检查集群健康状态
curl -X GET "http://localhost:9200/_cluster/health?pretty"# 查看节点信息
curl -X GET "http://localhost:9200/_cat/nodes?v"# 查看索引信息
curl -X GET "http://localhost:9200/_cat/indices?v"# 查看分片信息
curl -X GET "http://localhost:9200/_cat/shards?v"# 查看集群统计
curl -X GET "http://localhost:9200/_cluster/stats?pretty"

2.2 索引操作

bash

# 创建索引
curl -X PUT "http://localhost:9200/user_index" -H 'Content-Type: application/json' -d'
{"settings": {"number_of_shards": 3,"number_of_replicas": 1},"mappings": {"properties": {"id": {"type": "long"},"name": {"type": "text"},"age": {"type": "integer"},"email": {"type": "keyword"},"createTime": {"type": "date"},"address": {"type": "text"}}}
}'# 查看索引映射
curl -X GET "http://localhost:9200/user_index/_mapping?pretty"# 删除索引
curl -X DELETE "http://localhost:9200/user_index"

2.3 文档操作

bash

# 添加文档
curl -X POST "http://localhost:9200/user_index/_doc/1" -H 'Content-Type: application/json' -d'
{"id": 1,"name": "张三","age": 25,"email": "zhangsan@example.com","createTime": "2023-01-01T10:00:00","address": "北京市朝阳区"
}'# 获取文档
curl -X GET "http://localhost:9200/user_index/_doc/1?pretty"# 更新文档
curl -X POST "http://localhost:9200/user_index/_update/1" -H 'Content-Type: application/json' -d'
{"doc": {"age": 26}
}'# 删除文档
curl -X DELETE "http://localhost:9200/user_index/_doc/1"

2.4 搜索操作

bash

# 简单搜索
curl -X GET "http://localhost:9200/user_index/_search?q=name:张三&pretty"# 复杂搜索
curl -X GET "http://localhost:9200/user_index/_search" -H 'Content-Type: application/json' -d'
{"query": {"bool": {"must": [{"match": {"name": "张三"}}],"filter": [{"range": {"age": {"gte": 20, "lte": 30}}}]}},"sort": [{"createTime": {"order": "desc"}}],"from": 0,"size": 10
}'

3. Spring Boot集成

3.1 Maven依赖

xml

7.17.10
org.springframework.bootspring-boot-starter-data-elasticsearchorg.elasticsearch.clientelasticsearch-rest-high-level-client${elasticsearch.version}org.springframework.bootspring-boot-starter-web

3.2 配置文件

yaml

# application.yml
spring:elasticsearch:rest:uris: http://192.168.1.101:9200,http://192.168.1.102:9201,http://192.168.1.103:9202connection-timeout: 10sread-timeout: 30s# 自定义配置
app:elasticsearch:index:user: user_indexcluster:name: my-es-cluster

3.3 实体类

java

package com.example.esdemo.entity;import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;import java.time.LocalDateTime;@Data
@Document(indexName = "user_index", createIndex = false)
public class User {@Idprivate String id;@Field(type = FieldType.Long)private Long userId;@Field(type = FieldType.Text, analyzer = "ik_max_word")private String name;@Field(type = FieldType.Integer)private Integer age;@Field(type = FieldType.Keyword)private String email;@Field(type = FieldType.Date)private LocalDateTime createTime;@Field(type = FieldType.Text, analyzer = "ik_max_word")private String address;
}

3.4 Elasticsearch配置类

java

package com.example.esdemo.config;import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.elasticsearch.repository.config.EnableElasticsearchRepositories;import java.util.Arrays;
import java.util.Objects;@Configuration
@EnableElasticsearchRepositories(basePackages = "com.example.esdemo.repository")
public class ElasticsearchConfig {@Value("${spring.elasticsearch.rest.uris}")private String[] elasticsearchUris;@Bean(destroyMethod = "close")public RestHighLevelClient restHighLevelClient() {HttpHost[] hosts = Arrays.stream(elasticsearchUris).map(this::createHttpHost).filter(Objects::nonNull).toArray(HttpHost[]::new);return new RestHighLevelClient(RestClient.builder(hosts).setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder.setConnectTimeout(10000).setSocketTimeout(30000)));}private HttpHost createHttpHost(String uri) {try {String[] parts = uri.split(":");String scheme = parts[0];String host = parts[1].substring(2); // 去除"//"int port = Integer.parseInt(parts[2]);return new HttpHost(host, port, scheme);} catch (Exception e) {return null;}}
}

3.5 Repository接口

java

package com.example.esdemo.repository;import com.example.esdemo.entity.User;
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Repository;import java.util.List;@Repository
public interface UserRepository extends ElasticsearchRepository {// 根据姓名查询List findByName(String name);// 根据年龄范围查询List findByAgeBetween(Integer minAge, Integer maxAge);// 根据邮箱查询User findByEmail(String email);// 根据地址模糊查询List findByAddressContaining(String address);
}

3.6 服务层代码

java

package com.example.esdemo.service;import com.example.esdemo.entity.User;
import com.example.esdemo.repository.UserRepository;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Optional;@Slf4j
@Service
public class UserService {@Autowiredprivate UserRepository userRepository;@Autowiredprivate ElasticsearchRestTemplate elasticsearchRestTemplate;@Autowiredprivate RestHighLevelClient restHighLevelClient;/*** 添加用户*/public User addUser(User user) {user.setCreateTime(LocalDateTime.now());return userRepository.save(user);}/*** 根据ID查询用户*/public Optional getUserById(String id) {return userRepository.findById(id);}/*** 更新用户*/public User updateUser(User user) {if (user.getId() == null || !userRepository.existsById(user.getId())) {throw new RuntimeException("用户不存在");}return userRepository.save(user);}/*** 删除用户*/public void deleteUser(String id) {userRepository.deleteById(id);}/*** 批量添加用户*/public Iterable batchAddUsers(List users) {users.forEach(user -> user.setCreateTime(LocalDateTime.now()));return userRepository.saveAll(users);}/*** 根据姓名搜索用户*/public List searchByName(String name) {return userRepository.findByName(name);}/*** 复杂搜索：姓名 + 年龄范围*/public SearchHits complexSearch(String name, Integer minAge, Integer maxAge, int page, int size) {NativeSearchQuery query = new NativeSearchQueryBuilder().withQuery(QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("name", name)).filter(QueryBuilders.rangeQuery("age").gte(minAge).lte(maxAge))).withPageable(PageRequest.of(page, size)).build();return elasticsearchRestTemplate.search(query, User.class);}/*** 聚合查询：按年龄分组统计*/public void ageAggregation() throws IOException {SearchRequest searchRequest = new SearchRequest("user_index");SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();searchSourceBuilder.aggregation(AggregationBuilders.terms("age_group").field("age"));searchRequest.source(searchSourceBuilder);SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);Terms ageGroups = response.getAggregations().get("age_group");for (Terms.Bucket bucket : ageGroups.getBuckets()) {log.info("年龄: {}, 数量: {}", bucket.getKeyAsString(), bucket.getDocCount());}}
}

3.7 集群监控服务

java

package com.example.esdemo.service;import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsRequest;
import org.elasticsearch.action.admin.cluster.stats.ClusterStatsResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.cluster.health.ClusterHealthStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;@Slf4j
@Service
public class ClusterMonitorService {@Autowiredprivate RestHighLevelClient restHighLevelClient;/*** 获取集群健康状态*/public Map getClusterHealth() {Map healthInfo = new HashMap<>();try {ClusterHealthRequest request = new ClusterHealthRequest();ClusterHealthResponse response = restHighLevelClient.cluster().health(request, RequestOptions.DEFAULT);healthInfo.put("clusterName", response.getClusterName());healthInfo.put("status", response.getStatus().name());healthInfo.put("numberOfNodes", response.getNumberOfNodes());healthInfo.put("numberOfDataNodes", response.getNumberOfDataNodes());healthInfo.put("activeShards", response.getActiveShards());healthInfo.put("activePrimaryShards", response.getActivePrimaryShards());healthInfo.put("unassignedShards", response.getUnassignedShards());healthInfo.put("timedOut", response.isTimedOut());} catch (IOException e) {log.error("获取集群健康状态失败", e);healthInfo.put("error", e.getMessage());}return healthInfo;}/*** 获取集群统计信息*/public Map getClusterStats() {Map statsInfo = new HashMap<>();try {ClusterStatsRequest request = new ClusterStatsRequest();ClusterStatsResponse response = restHighLevelClient.cluster().stats(request, RequestOptions.DEFAULT);statsInfo.put("nodesCount", response.getNodesStats().getCount());statsInfo.put("indicesCount", response.getIndicesStats().getIndexCount());statsInfo.put("docsCount", response.getIndicesStats().getDocs().getCount());statsInfo.put("storeSize", response.getIndicesStats().getStore().getSize());statsInfo.put("jvmHeapUsed", response.getNodesStats().getJvm().getHeapUsed());statsInfo.put("jvmHeapMax", response.getNodesStats().getJvm().getHeapMax());} catch (IOException e) {log.error("获取集群统计信息失败", e);statsInfo.put("error", e.getMessage());}return statsInfo;}/*** 检查集群是否健康*/public boolean isClusterHealthy() {try {ClusterHealthRequest request = new ClusterHealthRequest();ClusterHealthResponse response = restHighLevelClient.cluster().health(request, RequestOptions.DEFAULT);return response.getStatus() != ClusterHealthStatus.RED;} catch (IOException e) {log.error("检查集群健康状态失败", e);return false;}}/*** 定时监控集群状态（每5分钟执行一次）*/@Scheduled(fixedRate = 300000)public void scheduledClusterMonitor() {Map healthInfo = getClusterHealth();Map statsInfo = getClusterStats();log.info("=== Elasticsearch集群监控 ===");log.info("集群状态: {}", healthInfo.get("status"));log.info("节点数量: {}", healthInfo.get("numberOfNodes"));log.info("索引数量: {}", statsInfo.get("indicesCount"));log.info("文档数量: {}", statsInfo.get("docsCount"));log.info("JVM堆内存使用: {}/{}", statsInfo.get("jvmHeapUsed"), statsInfo.get("jvmHeapMax"));if (!isClusterHealthy()) {log.warn("集群状态异常，请及时处理！");// 可以发送告警邮件或短信}}
}

3.8 控制器

java

package com.example.esdemo.controller;import com.example.esdemo.entity.User;
import com.example.esdemo.service.ClusterMonitorService;
import com.example.esdemo.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.web.bind.annotation.*;import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;@Slf4j
@RestController
@RequestMapping("/api/es")
public class ElasticsearchController {@Autowiredprivate UserService userService;@Autowiredprivate ClusterMonitorService clusterMonitorService;/*** 添加用户*/@PostMapping("/user")public User addUser(@RequestBody User user) {return userService.addUser(user);}/*** 根据ID查询用户*/@GetMapping("/user/{id}")public Optional getUser(@PathVariable String id) {return userService.getUserById(id);}/*** 更新用户*/@PutMapping("/user")public User updateUser(@RequestBody User user) {return userService.updateUser(user);}/*** 删除用户*/@DeleteMapping("/user/{id}")public void deleteUser(@PathVariable String id) {userService.deleteUser(id);}/*** 搜索用户*/@GetMapping("/user/search")public SearchHits searchUsers(@RequestParam String name,@RequestParam(defaultValue = "0") Integer minAge,@RequestParam(defaultValue = "100") Integer maxAge,@RequestParam(defaultValue = "0") Integer page,@RequestParam(defaultValue = "10") Integer size) {return userService.complexSearch(name, minAge, maxAge, page, size);}/*** 获取集群健康状态*/@GetMapping("/cluster/health")public Map getClusterHealth() {return clusterMonitorService.getClusterHealth();}/*** 获取集群统计信息*/@GetMapping("/cluster/stats")public Map getClusterStats() {return clusterMonitorService.getClusterStats();}/*** 检查集群是否健康*/@GetMapping("/cluster/healthy")public Map isClusterHealthy() {Map result = new HashMap<>();result.put("healthy", clusterMonitorService.isClusterHealthy());return result;}
}

3.9 启动类

java

package com.example.esdemo;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;@SpringBootApplication
@EnableScheduling
public class EsDemoApplication {public static void main(String[] args) {SpringApplication.run(EsDemoApplication.class, args);}
}

4. 测试和验证

4.1 创建测试数据

java

@Component
public class DataInitializer implements CommandLineRunner {@Autowiredprivate UserService userService;@Overridepublic void run(String... args) throws Exception {// 添加测试数据User user1 = new User();user1.setUserId(1L);user1.setName("张三");user1.setAge(25);user1.setEmail("zhangsan@example.com");user1.setAddress("北京市朝阳区");User user2 = new User();user2.setUserId(2L);user2.setName("李四");user2.setAge(30);user2.setEmail("lisi@example.com");user2.setAddress("上海市浦东新区");userService.addUser(user1);userService.addUser(user2);log.info("测试数据初始化完成");}
}

4.2 测试API调用

bash

# 检查集群健康状态
curl http://localhost:8080/api/es/cluster/health# 添加用户
curl -X POST http://localhost:8080/api/es/user \-H "Content-Type: application/json" \-d '{"userId": 3, "name": "王五", "age": 28, "email": "wangwu@example.com", "address": "广州市天河区"}'# 搜索用户
curl "http://localhost:8080/api/es/user/search?name=张三&minAge=20&maxAge=30"

这个完整的示例涵盖了Elasticsearch 7.17的安装、集群配置、客户端操作和Spring Boot集成，包括监控功能。所有代码都兼容JDK 1.8。