elasticsearch与mysql数据同步多个表（logstash）

Posted

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了elasticsearch与mysql数据同步多个表（logstash）相关的知识，希望对你有一定的参考价值。

参考技术A products索引字段展示

categorys索引字段展示

配置文件内容展示(同时同步products和categorys)

attribute索引字段展示

products索引字段展示

需要给两个txt文件相应的权限，详见 单表操作

同步后的products索引数据

elasticsearch与mysql数据同步

目录

数据同步

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

一.思路分析

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

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

1.同步调用

方案一：同步调用

基本步骤如下：

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

2.异步通知

方案二：异步通知

流程如下：

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

3.监听binlog

方案三：监听binlog

流程如下：

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

4.选择

方式一：同步调用

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

方式二：异步通知

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

方式三：监听binlog

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

二.实现数据同步

1.思路

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

步骤：

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

  • 声明exchange、queue、RoutingKey

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

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

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

2.导入demo

资料：资料在这里

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

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

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

3.声明交换机、队列

MQ结构如图：

3.1引入依赖

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

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

3.2 配置文件

spring:
  rabbitmq:
    host: 192.168.1.100
    username: guest
    password: guest
    virtual-host: /

3.3 声明队列交换机名称

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

package cn.itcast.hotel.constants;

    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.4 声明队列交换机

在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);

4.发送MQ消息

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

4.1 事务配置类

保证Rabbitmq在提交事务后执行。事务控制的详情

package cn.itcast.hotel.config;

import com.sun.istack.internal.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Component("afterCommitExecutor")
public class AfterCommitExecutor extends TransactionSynchronizationAdapter implements Executor 
    private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<List<Runnable>>();
    private ThreadPoolExecutor threadPool;

    private Logger logger = LoggerFactory.getLogger(AfterCommitExecutor.class);
    
    @PostConstruct
    public void init() 
        logger.debug("初始化线程池。。。");
        int availableProcessors = Runtime.getRuntime().availableProcessors();
        if (0 >= availableProcessors) 
            availableProcessors = 1;
        
        int maxPoolSize = (availableProcessors > 5) ? availableProcessors * 2 : 5;
        logger.debug("CPU Processors :%s MaxPoolSize:%s", availableProcessors, maxPoolSize);
        threadPool = new ThreadPoolExecutor(
            availableProcessors,
            maxPoolSize,
            60,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(maxPoolSize * 2),
            Executors.defaultThreadFactory(),
            new RejectedExecutionHandler() 
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) 
                    logger.debug("Task:%s rejected", r.toString());
                    if (!executor.isShutdown()) 
                        executor.getQueue().poll();
                        executor.execute(r);
                    
                
            
        );
    

    @PreDestroy
    public void destroy() 
        logger.debug("销毁线程池。。。");
        if (null != threadPool && !threadPool.isShutdown()) 
            threadPool.shutdown();
        
    

    @Override
    public void execute(@NotNull Runnable runnable) 
        if (!TransactionSynchronizationManager.isSynchronizationActive()) 
            runnable.run();
            return;
        
        List<Runnable> threadRunnables = RUNNABLES.get();
        if (threadRunnables == null) 
            threadRunnables = new ArrayList<Runnable>();
            RUNNABLES.set(threadRunnables);
            TransactionSynchronizationManager.registerSynchronization(this);
        
        threadRunnables.add(runnable);
    

    @Override
    public void afterCommit() 
        logger.debug("事务提交完成处理 ... ");
        List<Runnable> threadRunnables = RUNNABLES.get();
        for (int i = 0; i < threadRunnables.size(); i++) 
            Runnable runnable = threadRunnables.get(i);
            try 
                threadPool.execute(runnable);
             catch (RuntimeException e) 
                logger.error("", e);
            
        
    

    @Override
    public void afterCompletion(int status) 
        logger.debug("事务处理完毕 .... ");
        RUNNABLES.remove();

4.2 service 代码

 @Override
    @Transactional
    public boolean save(Hotel hotel) 
        logger.info("----- into  insert service -----");
        hotel.setId(Long.getLong(UUID.randomUUID().toString()));
        int i = hotelMapper.insert(hotel);
        afterCommitExecutor.execute(new Runnable() 
            @Override
            public void run() 
                rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE,MqConstants.HOTEL_INSERT_KEY,hotel.getId().toString());
                logger.info("----- rabbitmq send message -----");
            
        );


        if (1==i ||"1".equals(1) )
            return true;

        else 
            return false;
        
    

  
    @Override
    @Transactional
    public boolean updateById(Hotel hotel) 
        logger.info("----- into service -----");

        //修改DB的数据
        int i = hotelMapper.updateById(hotel);

        // 使用AfterCommitExecutor
        afterCommitExecutor.execute(new Runnable() 
            @Override
            public void run() 
                rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_INSERT_KEY, hotel.getId().toString());
                logger.info("----- rabbitmq send message -----");
            
        );

        logger.info("--------- out service ----------");

        if (1==i ||"1".equals(1) )
            return true;

        else 
            return false;
        
    

    @Override
    public void removeById(Long id) 
        logger.info("----- into service -----");

        //修改DB的数据
        int i = hotelMapper.deleteById(id);

        // 使用AfterCommitExecutor
        afterCommitExecutor.execute(new Runnable() 
            @Override
            public void run() 
                rabbitTemplate.convertAndSend(MqConstants.HOTEL_EXCHANGE, MqConstants.HOTEL_DELETE_KEY, id);
                logger.info("----- rabbitmq send message -----");
            
        );

        logger.info("--------- out service ----------");

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.发送请求
        restHighLevelClient.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.发送请求
        restHighLevelClient.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)MySQL数据同步到ES集群(MySQL数据库与ElasticSearch全文检索的同步)


 elasticsearch与mysql数据同步


 20.elasticsearch-jdbc实现MySQL同步到ElasticSearch(ES与关系型数据库同步)


 [es和数据库怎么同步]mysql与elasticsearch实时同步常用插件及优缺点对比(ES与关系型数据库同步)


 Elasticsearch学习笔记-p8（实现elasticsearch与mysql的数据同步）


 Mysql数据同步到Elasticsearch