ES数据同步&集群

Posted ?Suki

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ES数据同步&集群相关的知识，希望对你有一定的参考价值。

【ES】数据同步&集群

3.数据同步

elasticsearch中的酒店数据来自于mysql数据库，因此mysql数据发生改变时，elasticsearch也必须跟着改变，这个就是elasticsearch与mysql之间的数据同步。

3.1.思路分析

常见的数据同步方案有三种：

  • 同步调用
  • 异步通知
  • 监听binlog

3.1.1.同步调用

方案一：同步调用

基本步骤如下：

  • hotel-demo对外提供接口，用来修改elasticsearch中的数据
  • 酒店管理服务在完成数据库操作后，直接调用hotel-demo提供的接口，

3.1.2.异步通知

方案二：异步通知

流程如下：

  • hotel-admin对mysql数据库数据完成增、删、改后，发送MQ消息
  • hotel-demo监听MQ，接收到消息后完成elasticsearch数据修改

3.1.3.监听binlog

方案三：监听binlog

流程如下：

  • 给mysql开启binlog功能
  • mysql完成增、删、改操作都会记录在binlog中
  • hotel-demo基于canal监听binlog变化，实时更新elasticsearch中的内容

3.1.4.选择

方式一：同步调用

  • 优点：实现简单，粗暴
  • 缺点：业务耦合度高

方式二：异步通知

  • 优点：低耦合，实现难度一般
  • 缺点：依赖mq的可靠性

方式三：监听binlog

  • 优点：完全解除服务间耦合
  • 缺点：开启binlog增加数据库负担、实现复杂度高

3.2.实现数据同步

3.2.1.思路

利用课前资料提供的hotel-admin项目作为酒店管理的微服务。当酒店数据发生增、删、改时，要求对elasticsearch中数据也要完成相同操作。

步骤：

  • 导入课前资料提供的hotel-admin项目，启动并测试酒店数据的CRUD

  • 声明exchange、queue、RoutingKey

  • 在hotel-admin中的增、删、改业务中完成消息发送

  • 在hotel-demo中完成消息监听，并更新elasticsearch中数据

  • 启动并测试数据同步功能

3.2.2.导入demo

导入课前资料提供的hotel-admin项目：

运行后，访问 http://localhost:8099

其中包含了酒店的CRUD功能：

3.2.3.声明交换机、队列

MQ结构如图：

1）引入依赖

在hotel-admin、hotel-demo中引入rabbitmq的依赖：

<!--amqp-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2）声明队列交换机名称

在hotel-admin和hotel-demo中的cn.itcast.hotel.constatnts包下新建一个类MqConstants：

package cn.itcast.hotel.constatnts;

    public class MqConstants 
    /**
     * 交换机
     */
    public final static String HOTEL_EXCHANGE = "hotel.topic";
    /**
     * 监听新增和修改的队列
     */
    public final static String HOTEL_INSERT_QUEUE = "hotel.insert.queue";
    /**
     * 监听删除的队列
     */
    public final static String HOTEL_DELETE_QUEUE = "hotel.delete.queue";
    /**
     * 新增或修改的RoutingKey
     */
    public final static String HOTEL_INSERT_KEY = "hotel.insert";
    /**
     * 删除的RoutingKey
     */
    public final static String HOTEL_DELETE_KEY = "hotel.delete";

3）声明队列交换机

在hotel-demo中，定义配置类，声明队列、交换机：

package cn.itcast.hotel.config;

import cn.itcast.hotel.constants.MqConstants;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MqConfig 
    @Bean
    public TopicExchange topicExchange()
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
    

    @Bean
    public Queue insertQueue()
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
    

    @Bean
    public Queue deleteQueue()
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
    

    @Bean
    public Binding insertQueueBinding()
        return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
    

    @Bean
    public Binding deleteQueueBinding()
        return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);

3.2.4.发送MQ消息

在hotel-admin中的增、删、改业务中分别发送MQ消息：

3.2.5.接收MQ消息

hotel-demo接收到MQ消息要做的事情包括：

  • 新增消息：根据传递的hotel的id查询hotel信息，然后新增一条数据到索引库
  • 删除消息：根据传递的hotel的id删除索引库中的一条数据

1）首先在hotel-demo的cn.itcast.hotel.service包下的IHotelService中新增新增、删除业务

void deleteById(Long id);

void insertById(Long id);

2）给hotel-demo中的cn.itcast.hotel.service.impl包下的HotelService中实现业务：

@Override
public void deleteById(Long id) 
    try 
        // 1.准备Request
        DeleteRequest request = new DeleteRequest("hotel", id.toString());
        // 2.发送请求
        client.delete(request, RequestOptions.DEFAULT);
     catch (IOException e) 
        throw new RuntimeException(e);
    


@Override
public void insertById(Long id) 
    try 
        // 0.根据id查询酒店数据
        Hotel hotel = getById(id);
        // 转换为文档类型
        HotelDoc hotelDoc = new HotelDoc(hotel);

        // 1.准备Request对象
        IndexRequest request = new IndexRequest("hotel").id(hotel.getId().toString());
        // 2.准备Json文档
        request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
        // 3.发送请求
        client.index(request, RequestOptions.DEFAULT);
     catch (IOException e) 
        throw new RuntimeException(e);

3）编写监听器

在hotel-demo中的cn.itcast.hotel.mq包新增一个类：

package cn.itcast.hotel.mq;

import cn.itcast.hotel.constants.MqConstants;
import cn.itcast.hotel.service.IHotelService;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class HotelListener 

    @Autowired
    private IHotelService hotelService;

    /**
     * 监听酒店新增或修改的业务
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
    public void listenHotelInsertOrUpdate(Long id)
        hotelService.insertById(id);
    

    /**
     * 监听酒店删除的业务
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
    public void listenHotelDelete(Long id)
        hotelService.deleteById(id);

4.集群

单机的elasticsearch做数据存储，必然面临两个问题：海量数据存储问题、单点故障问题。

  • 海量数据存储问题：将索引库从逻辑上拆分为N个分片（shard），存储到多个节点
  • 单点故障问题：将分片数据在不同节点备份（replica ）

ES集群相关概念:

  • 集群（cluster）：一组拥有共同的 cluster name 的 节点。

  • 节点（node) ：集群中的一个 Elasticearch 实例

  • 分片（shard）：索引可以被拆分为不同的部分进行存储，称为分片。在集群环境下，一个索引的不同分片可以拆分到不同的节点中

    解决问题：数据量太大，单点存储量有限的问题。

    此处，我们把数据分成3片：shard0、shard1、shard2

  • 主分片（Primary shard）：相对于副本分片的定义。

  • 副本分片（Replica shard）每个主分片可以有一个或者多个副本，数据和主分片一样。

    ​

数据备份可以保证高可用，但是每个分片备份一份，所需要的节点数量就会翻一倍，成本实在是太高了！

为了在高可用和成本间寻求平衡，我们可以这样做：

  • 首先对数据分片，存储到不同节点
  • 然后对每个分片进行备份，放到对方节点，完成互相备份

这样可以大大减少所需要的服务节点数量，如图，我们以3分片，每个分片备份一份为例：

现在，每个分片都有1个备份，存储在3个节点：

  • node0：保存了分片0和1
  • node1：保存了分片0和2
  • node2：保存了分片1和2

4.1.搭建ES集群

参考课前资料的文档：

其中的第四章节：

4.2.集群脑裂问题

4.2.1.集群职责划分

elasticsearch中集群节点有不同的职责划分：

默认情况下，集群中的任何一个节点都同时具备上述四种角色。

但是真实的集群一定要将集群职责分离：

  • master节点：对CPU要求高，但是内存要求第
  • data节点：对CPU和内存要求都高
  • coordinating节点：对网络带宽、CPU要求高

职责分离可以让我们根据不同节点的需求分配不同的硬件去部署。而且避免业务之间的互相干扰。

一个典型的es集群职责划分如图：

4.2.2.脑裂问题

脑裂是因为集群中的节点失联导致的。

例如一个集群中，主节点与其它节点失联：

此时，node2和node3认为node1宕机，就会重新选主：

当node3当选后，集群继续对外提供服务，node2和node3自成集群，node1自成集群，两个集群数据不同步，出现数据差异。

当网络恢复后，因为集群中有两个master节点，集群状态的不一致，出现脑裂的情况：

解决脑裂的方案是，要求选票超过 ( eligible节点数量 + 1 ）/ 2 才能当选为主，因此eligible节点数量最好是奇数。对应配置项是discovery.zen.minimum_master_nodes，在es7.0以后，已经成为默认配置，因此一般不会发生脑裂问题

例如：3个节点形成的集群，选票必须超过 （3 + 1） / 2 ，也就是2票。node3得到node2和node3的选票，当选为主。node1只有自己1票，没有当选。集群中依然只有1个主节点，没有出现脑裂。

4.2.3.小结

master eligible节点的作用是什么？

  • 参与集群选主
  • 主节点可以管理集群状态、管理分片信息、处理创建和删除索引库的请求

data节点的作用是什么？

  • 数据的CRUD

coordinator节点的作用是什么？

  • 路由请求到其它节点

  • 合并查询到的结果，返回给用户

4.3.集群分布式存储

当新增文档时，应该保存到不同分片，保证数据均衡，那么coordinating node如何确定数据该存储到哪个分片呢？

4.3.1.分片存储测试

插入三条数据：

测试可以看到，三条数据分别在不同分片：

结果：

4.3.2.分片存储原理

elasticsearch会通过hash算法来计算文档应该存储到哪个分片：

说明：

  • _routing默认是文档的id
  • 算法与分片数量有关，因此索引库一旦创建，分片数量不能修改！

新增文档的流程如下：

解读：

  • 1）新增一个id=1的文档
  • 2）对id做hash运算，假如得到的是2，则应该存储到shard-2
  • 3）shard-2的主分片在node3节点，将数据路由到node3
  • 4）保存文档
  • 5）同步给shard-2的副本replica-2，在node2节点
  • 6）返回结果给coordinating-node节点

4.4.集群分布式查询

elasticsearch的查询分成两个阶段：

  • scatter phase：分散阶段，coordinating node会把请求分发到每一个分片

  • gather phase：聚集阶段，coordinating node汇总data node的搜索结果，并处理为最终结果集返回给用户

4.5.集群故障转移

集群的master节点会监控集群中的节点状态，如果发现有节点宕机，会立即将宕机节点的分片数据迁移到其它节点，确保数据安全，这个叫做故障转移。

1）例如一个集群结构如图：

现在，node1是主节点，其它两个节点是从节点。

2）突然，node1发生了故障：

宕机后的第一件事，需要重新选主，例如选中了node2：

node2成为主节点后，会检测集群监控状态，发现：shard-1、shard-0没有副本节点。因此需要将node1上的数据迁移到node2、node3：

学习笔记 from 黑马程序员

By – Suki 2023/4/9

Elasticsearch——Windows下ES集群部署 & Linux下ES单节点集群部署

1.开篇

在之前的两篇文章中，说白了就是在windows下部署的ES单节点的环境。

这篇文章主要是说一下windows下部署ES集群、Linux下单节点部署。

单台 Elasticsearch 服务器提供服务，往往都有最大的负载能力，超过这个阈值，服务器性能就会大大降低甚至不可用，所以生产环境中，一般都是运行在指定服务器集群中。除了负载能力，单点服务器也存在其他问题：

  • 单台机器存储容量有限
  • 单服务器容易出现单点故障，无法实现高可用
  • 单服务的并发处理能力有限

配置服务器集群时，集群中节点数量没有限制，大于等于 2 个节点就可以看做是集群了。一般出于高性能及高可用方面来考虑集群中节点数量都是 3 个以上。

一个集群就是由一个或多个服务器节点组织在一起，共同持有整个的数据，并一起提供索引和搜索功能。一个 Elasticsearch 集群有一个唯一的名字标识，这个名字默认就是”elasticsearch”。这个名字是重要的，因为一个节点只能通过指定某个集群的名字，来加入这个集群。

集群中包含很多服务器，一个节点就是其中的一个服务器。作为集群的一部分，它存储数据，参与集群的索引和搜索功能。一个节点也是由一个名字来标识的，默认情况下，这个名字是一个随机的漫威漫画角色的名字，这个名字会在启动的时候赋予节点。这个名字对于管理工作来说挺重要的，因为在这个管理过程中，你会去确定网络中的哪些服务器对应于 Elasticsearch 集群中的哪些节点。一个节点可以通过配置集群名称的方式来加入一个指定的集群。默认情况下，每个节点都会被安排加入到一个叫做“elasticsearch”的集群中，这意味着，如果你在你的网络中启动了若干个节点，并假定它们能够相互发现彼此，它们将会自动地形成并加入到一个叫做“elasticsearch”的集群中。

2.Windows部署ES集群

首先创建一个elasticsearch-cluster文件夹，在其中创建三个一模一样的由ES安装包解压之后的文件。

同时在每一个节点的config目录下的elasticsearch.yml 配置文件中做如下修改：

#集群名称，节点之间要保持一致
cluster.name: my-elasticsearch

#节点名称，集群内要唯一
node.name: node-1001
node.master: true
node.data: true

#ip地址
network.host: localhost

#http端口
http.port: 1001
#tcp监听端口
transport.tcp.port: 9301

#跨域配置
#action.destructive_requires_name: true
http.cors.enabled: true
http.cors.allow-origin: "*"
#集群名称，节点之间要保持一致
cluster.name: my-elasticsearch

#节点名称，集群内要唯一
node.name: node-1002
node.master: true
node.data: true

#ip地址
network.host: localhost
#http端口
http.port: 1002
#tcp监听端口
transport.tcp.port: 9302

#候选主节点的地址，在开启服务后可以被选为主节点
discovery.seed_hosts: ["localhost:9301"]
discovery.zen.fd.ping_timeout: 1m
discovery.zen.fd.ping_retries: 5

#跨域配置
#action.destructive_requires_name: true
http.cors.enabled: true
http.cors.allow-origin: "*"

#集群名称，节点之间要保持一致
cluster.name: my-elasticsearch

#节点名称，集群内要唯一
node.name: node-1003
node.master: true
node.data: true

#ip地址
network.host: localhost
#http端口
http.port: 1003
#tcp监听端口
transport.tcp.port: 9303

#候选主节点的地址，在开启服务后可以被选为主节点
discovery.seed_hosts: ["localhost:9301","localhost:9302"]
discovery.zen.fd.ping_timeout: 1m
discovery.zen.fd.ping_retries: 5

#跨域配置
#action.destructive_requires_name: true
http.cors.enabled: true
http.cors.allow-origin: "*"

以上三个节点都配置完成之后，依次进入这三个节点的bin目录下，双击elasticsearch.bat，启动ES服务。

三个节点都已经启动，可以到Postman中检测一下集群节点的状况。

在节点1001中，创建索引之后，在1002、1003节点中均可查看。

3.Linux部署ES单节点

首先直接rz，将es的linux版本压缩包上传至linux中，然后tar -zxvf解压。

由于解压之后，文件名太长，我这里先做了一次改名。

因为安全问题，Elasticsearch 不允许 root 用户直接运行。所以要创建新用户，我就在 root 用户中创建新用户es。

chown -R es:es /opt/module/es #文件夹所有者。

之后切换到es用户中，bin目录下的命令直接启动即可。

修改 /opt/module/es/config/elasticsearch.yml 文件
# 加入如下配置
cluster.name: elasticsearch
node.name: node-1
network.host: 0.0.0.0
http.port: 9200
cluster.initial_master_nodes: ["node-1"]

修改/etc/security/limits.conf
# 在文件末尾中增加下面内容
# 每个进程可以打开的文件数的限制
es soft nofile 65536
es hard nofile 65536

修改/etc/security/limits.d/20-nproc.conf
# 在文件末尾中增加下面内容
# 每个进程可以打开的文件数的限制
es soft nofile 65536
es hard nofile 65536 

# 操作系统级别对每个用户创建的进程数的限制

* hard nproc 4096
# 注：* 带表 Linux 所有用户名称

修改/etc/sysctl.conf
# 在文件中增加下面内容
# 一个进程可以拥有的 VMA(虚拟内存区域)的数量,默认值为 65536
vm.max_map_count=655360

重新加载
sysctl -p

修改完成之后，启动。

启动完成，到Poatman中测试。

4.Linux部署ES集群

最后简单的说一下Linux下ES集群部署，其实也不难，只是我的虚拟机内存性能等各种问题限制，所以这里无法实操了。

大概的步骤就是：开启3个Linux虚拟机终端（ip不能一样），分别在3个Linux终端下解压ES的安装包，之后在3个Linux终端中创建用户es，按照上面单节点方式进行多个配置文件的修改。

修改/opt/module/es/config/elasticsearch.yml 文件

# 加入如下配置
#集群名称
cluster.name: cluster-es
#节点名称，每个节点的名称不能重复
node.name: node-1
#ip 地址，每个节点的地址不能重复
network.host: linux1
#是不是有资格主节点
node.master: true
node.data: true
http.port: 9200
# head 插件需要这打开这两个配置
http.cors.allow-origin: "*"
http.cors.enabled: true
http.max_content_length: 200mb
#es7.x 之后新增的配置，初始化一个新的集群时需要此配置来选举 master
cluster.initial_master_nodes: ["node-1"]
#es7.x 之后新增的配置，节点发现
discovery.seed_hosts: ["linux1:9300","linux2:9300","linux3:9300"]
gateway.recover_after_nodes: 2
network.tcp.keep_alive: true
network.tcp.no_delay: true
transport.tcp.compress: true
#集群内同时启动的数据任务个数，默认是 2 个
cluster.routing.allocation.cluster_concurrent_rebalance: 16
#添加或删除节点及负载均衡时并发恢复的线程个数，默认 4 个
cluster.routing.allocation.node_concurrent_recoveries: 16
#初始化数据恢复时，并发恢复线程的个数，默认 4 个
cluster.routing.allocation.node_initial_primaries_recoveries: 16

修改/etc/security/limits.conf

# 在文件末尾中增加下面内容
es soft nofile 65536
es hard nofile 65536

修改/etc/security/limits.d/20-nproc.conf

# 在文件末尾中增加下面内容
es soft nofile 65536
es hard nofile 65536
* hard nproc 4096
# 注：* 带表 Linux 所有用户名称

修改/etc/sysctl.conf

# 在文件中增加下面内容
vm.max_map_count=655360

重新加载：sysctl -p

最后分别在3个Linux终端下，切换到es用户，启动配置好的ES节点。（如果启动失败，重新切回root用户，在3个Linux终端下分别执行一次这个命令：

chown -R es:es /opt/es-cluster-node1、chown -R es:es /opt/es-cluster-node2、chown -R es:es /opt/es-cluster-node3 ）

#启动
bin/elasticsearch
#后台启动
bin/elasticsearch -d

启动完成之后，测试：http://虚拟机ip:9200/_cat/nodes，即可看到和上面windows集群在Poatman中相同的测试结果。

以上是关于ES数据同步&集群的主要内容，如果未能解决你的问题，请参考以下文章

K8S 日志收集：ES 集群安装

Python Elasticsearch API操作ES集群

MySQL数据同步到ES集群(MySQL数据库与ElasticSearch全文检索的同步)

docker-compose安装elasticsearch集群

Elasticsearch单机安装_集群搭建_索引基本操作_Head插件安装与基本操作_ik分词器配置_logstash实现数据同步_教程

es集群搭建