关键词：
目录
数据同步
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)查看详情
elasticsearch与mysql数据同步多个表（logstash）
参考技术Aproducts索引字段展示categorys索引字段展示配置文件内容展示(同时同步products和categorys)attribute索引字段展示products索引字段展示需要给两个txt文件相应的权限，详见单表操作同步后的products索引数据 查看详情
elasticsearch与mysql数据同步(代码片段)
...送MQ消息4.1事务配置类4.2service代码5.接收MQ消息数据同步elasticsearch中的酒店数据来自于mysql数据库，因此mysql数据发生改变时，elasticsearch也必须跟着改变，这个就是elasticsearch与mysql之间的数据同步。一.思路分析常见的... 查看详情
mysql数据同步到elasticsearch(代码片段)
〝古人学问遗无力，少壮功夫老始成〞要通过elasticsearch实现数据检索，首先要将mysql中的数据导入elasticsearch，并实现数据源与elasticsearch数据同步，这里使用的数据源是Mysql数据库，目前mysql与elasticsearch常用的... 查看详情
mysql数据同步到elasticsearch(代码片段)
〝古人学问遗无力，少壮功夫老始成〞要通过elasticsearch实现数据检索，首先要将mysql中的数据导入elasticsearch，并实现数据源与elasticsearch数据同步，这里使用的数据源是Mysql数据库，目前mysql与elasticsearch常用的... 查看详情
canal+rocketmq实现mysql与elasticsearch数据同步(代码片段)
1.引言在很多业务情况下，我们都会在系统中引入ElasticSearch搜索引擎作为做全文检索的优化方案。如果数据库数据发生更新，这时候就需要在业务代码中写一段同步更新ElasticSearch的代码。这种数据同步的代码跟业务代码... 查看详情
mysql数据同步至elasticsearch的相关实现方法
Python： MySQL数据同步到ES集群(MySQL数据库与ElasticSearch全文检索的同步)通过logstash将mysql数据同步至es中Springboot+ElasticSearch构建博客检索系统-学习笔记01Springboot+ElasticSearch构建博客检索系统-学习笔记02P43 43.新闻案例-数据... 查看详情
elasticsearch-jdbc批量同步mysql数据失败
...直接从ES中去查询想要的结果。通过一番查找，决定使用elasticsearch-jdbc进行数据的同步，五六张表的连接结果，在数据量小的开发与测试环境运行正常，但在数据量比较大的性能测试环境做数据同步的话就会出现问题 查看详情
canal:mysql数据同步elasticsearch，以高德地图为例
源代码与介绍：https://gitee.com/davidyem/spring-boot-data-syn 查看详情
数据中间件如何与mysql数据同步？(代码片段)
...3.监控binlog实现"同步"更新4.总结1.引入先前介绍了ElasticSearch，以及ES配合MySQL的问题，这种方案是让ES上的数据根据MySQL的数据做对照从而形成对应的索引，再将数据通过处理和封装存放在ES当中。（可回顾... 查看详情
elasticsearch-jdbc实现mysql同步到elasticsearch深入详解(代码片段)
Elasticsearch最少必要知识实战教程直播回放1.如何实现mysql与elasticsearch的数据同步？逐条转换为json显然不合适，需要借助第三方工具或者自己实现。核心功能点：同步增、删、改、查同步。2、mysql与elasticsearch同步的方... 查看详情
使用go-mysql-elasticsearch同步mysql数据库信息到elasticsearch(代码片段)
本文介绍如何使用go-mysql-elasticsearch同步mysql数据库信息到ElasticSearch。1.go-mysql-elasticsearch简介go-mysql-elasticsearch是一个将MySQL数据自动同步到Elasticsearch的服务。它首先使用mysqldump获取原始数据，然后用binlog增量地同步数据。github地... 查看详情
elasticsearch推荐一个同步mysql数据到elasticsearch的工具
1.概述转载：https://elasticsearch.cn/article/756 查看详情
使用logstash和jdbc确保elasticsearch与关系型数据库保持同步
为了充分利用Elasticsearch提供的强大搜索功能，很多公司都会在既有关系型数据库的基础上再部署Elasticsearch。在这种情况下，很可能需要确保Elasticsearch与所关联关系型数据库中的数据保持同步。因此，在本篇博文中，我会演示如... 查看详情
datax数据同步（mysql-->elasticsearch）
参考技术A链接：https://pan.baidu.com/s/1YthL24An_972MRAEPewH9A提取码：hxan安装过程自行百度链接：https://pan.baidu.com/s/1rUEkE3xcFQH3uZUoOgKTDg提取码：gwhx解压即可"column":["sgiidid","sgiid","ggoodsCode","goodsCode",&quo... 查看详情
实现mysql与es的增量数据同步
参照网上的文章:https://zhuanlan.zhihu.com/p/521914517一步步做实现增量同步思路是:将mysql增删改操作产生的binlog通过canal中间件增量同步到ES中一开始犯贱使用的版本是：mysql8.0、canal1.1.6、Elasticsearch 查看详情
elasticsearch同步mysql
ElasticSearch同步Mysql的插件选择了elasticsearch-jdbc,理由是活跃度高，持续更新,最新版本兼容elasticsearch-2.3.3.一、下载下载地址：https://github.com/jprante/elasticsearch-jdbc下载后解压,里面有bin、lib2个目录.二、mysql配置确保mysql能用，在mysql... 查看详情
logstash同步mysql数据到elasticsearch(代码片段)
目录1MySql数据到Elasticsearch1.1下载logstash1.2解压logstash1.3在logstash目录创建mysql文件夹1.4将mysql驱动文件和数据库查询文件放进mysql中1.5在config目录下创建mysqltoes.conf文件1.6mysqltoes.conf配置1.7启动logstash2配置语法讲解3启动方式4filebeat基... 查看详情