价格划算的东莞建网站公司,昆山专业做网站,wordpress添加继续阅读,福州医疗网站建设5.5 课程信息索引同步
5.5.1 技术方案
通过向索引中添加课程信息最终实现了课程的搜索#xff0c;我们发现课程信息是先保存在关系数据库中#xff0c;而后再写入索引#xff0c;这个过程是将关系数据中的数据同步到elasticsearch索引中的过程#xff0c;可以简单成为索引…5.5 课程信息索引同步
5.5.1 技术方案
通过向索引中添加课程信息最终实现了课程的搜索我们发现课程信息是先保存在关系数据库中而后再写入索引这个过程是将关系数据中的数据同步到elasticsearch索引中的过程可以简单成为索引同步。
通常项目中使用elasticsearch需要完成索引同步索引同步的方法很多
1、针对实时性非常高的场景需要满足数据的及时同步可以同步调用或使用Canal去实现。
1同步调用即在向MySQL写数据后远程调用搜索服务的接口写入索引此方法简单但是耦合代码太高。
2可以使用一个中间的软件canal解决耦合性的问题但存在学习与维护成本。
canal主要用途是基于 MySQL 数据库增量日志解析并能提供增量数据订阅和消费实现将MySQL的数据同步到消息队列、Elasticsearch、其它数据库等应用场景十分丰富。 它的地址
github地址https://github.com/alibaba/canal
版本下载地址https://github.com/alibaba/canal/releases
文档地址https://github.com/alibaba/canal/wiki/Docker-QuickStart Canal基于mysql的binlog技术实现数据同步什么是binlog它是一个文件二进制格式记录了对数据库更新的SQL语句向数据库写数据的同时向binlog文件里记录对应的sql语句。当数据库服务器发生了故障就可以使用binlog文件对数据库进行恢复。
所以使用canal是需要开启mysql的binlog写入功能Canal工作原理如下 1、canal 模拟 MySQL slave 的交互协议伪装自己为 MySQL slave 向 MySQL master 发送dump
协议
2、MySQL master 收到 dump 请求开始推送 binary log 给 slave (即 canal )
3、canal 解析 binary log 对象(原始为 byte 流)
详细使用Canal进行索引同步的步骤参考Canal实现索引同步.pdf 2、当索引同步的实时性要求不高时可用的技术比较多比如MQ、Logstash、任务调度等。
MQ向mysql写数据的时候向mq写入消息搜索服务监听MQ收到消息后写入索引。使用MQ的优势是代码解耦但是需要处理消息可靠性的问题有一定的技术成本做到消息可靠性需要做到生产者投递成功、消息持久化以及消费者消费成功三个方面另外还要做好消息幂等性问题。但是MQ会产生数据可靠性问题在之前的文档之我们讲解了MQ的消息可靠性。使用MQ技术实现难度还是比较大的。
Logstash 开源实时日志分析平台 ELK包括Elasticsearch、Kibana、LogstashLogstash负责收集、解析和转换日志信息可以实现MySQL与Elasticsearch之间的数据同步。也可以实现解耦合并且是官方推荐但需要增加学习与维护成本。
任务调度向mysql写数据的时候记录修改记录开启一个定时任务根据修改记录将数据同步到Elasticsearch。
我们实现这个同步功能使用索引同步的方案进行解决。
实现代码
1、拷贝CourseIndex 模型类到内容管理model 工程的dto包下。
2、在内容管理服务中添加FeignClient Java package com.xuecheng.content.feignclient; import com.xuecheng.content.model.dto.CourseIndex; import org.springframework.cloud.openfeign.FeignClient; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; /** * description 搜索服务远程接口 * author Mr.M * date 2022/9/20 20:29 * version 1.0 */ FeignClient(value search,fallbackFactory SearchServiceClientFallbackFactory.class) public interface SearchServiceClient { PostMapping(/search/index/course) public Boolean add(RequestBody CourseIndex courseIndex); }
定义SearchServiceClientFallbackFactory Java Slf4j Component public class SearchServiceClientFallbackFactory implements FallbackFactorySearchServiceClient { Override public SearchServiceClient create(Throwable throwable) { return new SearchServiceClient() { Override public Boolean add(CourseIndex courseIndex) { throwable.printStackTrace(); log.debug(调用搜索发生熔断走降级方法,熔断异常:, throwable.getMessage()); return false; } }; } } 3、编写课程索引任务执行方法
完善CoursePublishTask类中的saveCourseIndex方法 Java //保存课程索引信息 public void saveCourseIndex(MqMessage mqMessage,long courseId){ log.debug(保存课程索引信息,课程id:{},courseId); //消息id Long id mqMessage.getId(); //消息处理的service MqMessageService mqMessageService this.getMqMessageService(); //消息幂等性处理 int stageTwo mqMessageService.getStageTwo(id); if(stageTwo 0){ log.debug(课程索引已处理直接返回课程id:{},courseId); return ; } Boolean result saveCourseIndex(courseId); if(result){ //保存第一阶段状态 mqMessageService.completedStageTwo(id); } } private Boolean saveCourseIndex(Long courseId) { //取出课程发布信息 CoursePublish coursePublish coursePublishMapper.selectById(courseId); //拷贝至课程索引对象 CourseIndex courseIndex new CourseIndex(); BeanUtils.copyProperties(coursePublish,courseIndex); //远程调用搜索服务api添加课程信息到索引 Boolean add searchServiceClient.add(courseIndex); if(!add){ XueChengPlusException.cast(添加索引失败); } return add; }
