wordpress安装到子目录,广州优化防控措施,企业官方网站怎么做,htp免费域名注册网站为什么使用elasticsearch?数据很多么#xff1f;
项目使用Elasticsearch是实现了门户上对服务的搜索。
平台上的服务数据是并不是很多#xff0c;全国所有区域下的服务信息加一起几千条#xff0c;之所以使用Elasticsearch是因为#xff1a;
1、公司架构师在系统架构时…为什么使用elasticsearch?数据很多么
项目使用Elasticsearch是实现了门户上对服务的搜索。
平台上的服务数据是并不是很多全国所有区域下的服务信息加一起几千条之所以使用Elasticsearch是因为
1、公司架构师在系统架构时考虑几年后的数据及对全文检索使用的需求使用了Elasticsearch.
2、对服务信息进行搜索使用的是全文检索方式虽然MySQL也支持全文检索但是我们这个接口是面向 C端用户且对接口性能有要求所以使用了ES。
3、虽然现在数据量不大考虑几年后的数据量增长问题我们使用了Elasticsearch。
4、在项目中除了通过关键字搜索服务信息还有根据地理坐标进行搜索使用Elasticsearch也考虑了这一点。 实现目标
对于服务类型和服务项的名称进行索引当搜索服务项名称时展示出该服务项当搜索服务类型时展示出该类型下的所有服务项。
方案一 因为操作es和数据库 是分布式事务无法控制一致性。
方案二
使用CanalMQ Canal可与很多数据源进行对接将数据由MySQL同步到ES、MQ、DB等各个数据源。
Canal的意思是水道/管道/沟渠它相当于一个数据管道通过解析MySQL的binlog日志完成数据同步工作。 要理解上图中Canal的工作原理需要首先要知道MySQL主从数据同步的原理如下图 MySQL主从数据同步的原理
MySQL主从集群由MySQL主服务器(master)和MySQL从服务器(slave)组成MySQL主从数据同步是一种数据库复制技术进行写数据会先向主服务器写写成功后将数据同步到从服务器流程如下
1、主服务器将所有写操作INSERT、UPDATE、DELETE以二进制日志binlog的形式记录下来。
2、从服务器连接到主服务器发送dump 协议,请求获取主服务器上的binlog日志。
MySQL的dump协议是MySQL复制协议中的一部分。
3、MySQL master 收到 dump 请求开始推送 binary log 给 slave
4、从服务器解析日志根据日志内容更新从服务器的数据库完成从服务器的数据保持与主服务器同步。
理解了MySQL主从同步的原理Canal在整个过程充当什么角色呢
如下图 工作流程如下
1、Canal模拟 MySQL slave 的交互协议伪装自己为 MySQL slave 向 MySQL master 发送dump 协议
MySQL的dump协议是MySQL复制协议中的一部分。
2、MySQL master 收到 dump 请求开始推送 binary log 给 slave (即 canal )
。一旦连接建立成功Canal会一直等待并监听来自MySQL主服务器的binlog事件流当有新的数据库变更发生时MySQL master主服务器发送binlog事件流给Canal。
3、Canal会及时接收并解析这些变更事件并解析 binary log
通过以上流程可知Canal和MySQL master主服务器之间建立了长连接。 流程运营端人员修改信息canal监听到mysql的binlog日志并进行解析发送到mq,mq监听到消息 同步简历索引 MQ技术方案
目标
能说出如何保证MQ消息的可靠性
1保证生产消息可靠性
上述技术方案中有一个关键点首先数据增删改的信息是保证写入binlog的Canal解析出增删改的信息后写入MQ同步程序从MQ去读取消息如果MQ中的消息丢失了数据将无法进行同步。
如何保证MQ消息的可靠性
保证MQ消息的可靠性分两个方面保证生产消息的可靠性、保证消费消息的可靠性。
保证生产消息的可靠性
RabbitMQ提供生产者确认机制保证生产消息的可靠性技术方案如下 首先发送消息的方法如果执行失败会进行重试重试次数耗尽记录失败消息
如果重试失败则将数据保存到数据库中 通过MQ的提供的生产者确认机制保证生产消息的可靠性
使用生产者确认机制需要给每个消息指定一个唯一ID生产者确认机制通过异步回调的方式进行包括ConfirmCallback和Return回调。
ConfirmCallback消息发送到Broker会有一个结果返回给发送者表示消息是否处理成功
1消息成功投递到交换机返回ack
2消息未投递到交换机返回nack
在发送消息时指定回调对象 回调类中回调方法源代码
如果没有返回ack则将消息记录到失败消息表如果经过重试后返回了ack说明消息发送成功此时将消息从失败消息表删除。 Return回调如果消息发送到交换机成功了但是并没有到达队列此时会调用ReturnCallback回调方法在回调方法中我们可以收到失败的消息存入失败消息表以便进行补偿。
要使用Return回调需要开启设置
首先在shared-rabbitmq.yaml中配置rabbitMQ参数如下
spring:rabbitmq:publisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: true说明
publish-confirm-type开启publisher-confirm这里支持两种类型
simple同步等待confirm结果直到超时
correlated异步回调定义ConfirmCallbackMQ返回结果时会回调这个ConfirmCallback
publish-returns开启publish-return功能同样是基于callback机制不过是定义ReturnCallback
template.mandatory定义消息路由失败时的策略。true则调用ReturnCallbackfalse则直接丢弃消息 保证消费消息可靠性 首先设置消息持久化保证消息发送到MQ消息不丢失。具体需要设置交换机和队列支持持久化发送消息设置deliveryMode2。
RabbitMQ是通过消费者回执来确认消费者是否成功处理消息的消费者获取消息后应该向RabbitMQ发送ACK回执表明自己已经处理完成消息RabbitMQ收到ACK后删除消息。 spring:rabbitmq:....listener:simple:acknowledge-mode: auto #出现异常时返回nack消息回滚到mq没有异常返回ackretry:enabled: true # 开启消费者失败重试initial-interval: 1000 # 初识的失败等待时长为1秒multiplier: 10 # 失败的等待时长倍数下次等待时长 multiplier * last-intervalmax-attempts: 3 # 最大重试次数stateless: true # true无状态false有状态。如果业务中包含事务这里改为false能说出如何保证MQ幂等性或 如何防止重复消费 消费者在消费消息时难免出现重复消费的情况比如消费者没有向MQ返回ack导致重复消费所以消费者需要保证消费消息幂等性。
什么是幂等性
幂等性是指不论执行多少次其结果是一致的。
举例
收到消息需要向数据新增一条记录如果重复消费则会出现重复添加记录的问题。
下边根据场景分析解决方案
1、查询操作
本身具有幂等性。
2、添加操作
如果主键是自增则可能重复添加记录。
保证幂等性可以设置数据库的唯一约束比如添加学生信息将学号字段设置为唯一索引即使重复添加相同的学生同一个学号只会添加一条记录。
3、更新操作
如果是更新一个固定的值比如: update users set status 1 where id?本身具有幂等性。
如果只允许更新成功一次则可以使用token机制发送消息前生成一个token写入redis收到消息后解析出token从redis查询token如果成功则说明没有消费此时更新成功将token从redis删除当重复消费相同 的消息时由于token已经从redis删除不会再执行更新操作。
4、删除操作
与更新操作类似如果是删除某个具体的记录比如delete from users where id?本身具有幂等性。
如果只允许删除成功一次可以采用更新操作相同的方法。 可以百分百保证MQ的消息可靠性吗
保证消息可靠性分两个方面保证生产消息可靠性和保证消费消息可靠性。
保证生产消息可靠性
生产消息可靠性是通过判断MQ是否发送ack回执如果发nack表示发送消息失败此时会进行重发或记录到失败消息表通过定时任务进行补偿发送。如果Java程序并没有收到回执如jvm进程异常结束了或断电等因素此时将无法保证生产消息的可靠性。
保证消费消息可靠性
保证消费消息可靠性方案首先保证发送消息设置为持久化其次通过MQ的消费确认机制保证消费者消费成功消息后再将消息删除。
虽然设置了消息持久化消息进入MQ首先是在缓存存在MQ会根据一定的规则进行刷盘比如每隔几毫秒进行刷盘如果在消息还没有保存到磁盘时MQ进程终止此时将会丢失消息。虽然可以使用镜像队列用于在 RabbitMQ 集群中复制队列的消息这样做的目的是提高队列的可用性和容错性以防止在单个节点故障时导致消息的丢失。但也不能百分百保证消息不丢失。
虽然我们加了很多保证可靠性的机制这样也只是去提高消息的可靠性不能百分百做的可靠所以使用MQ的场景要考虑这种问题的存在做好补偿处理任务。
CanalMQ同步流程
实现将MySQL的变更数据通过Canal写入MQ。 配置CanalMQ数据同步环境
参考配置链接
Docshttps://mx67xggunk5.feishu.cn/wiki/Yifpw51Qoim81akHF7ic5ye3npd
根据CanalMQ同步流程下边进行如下配置 配置Mysql主从同步开启MySQL主服务器的binlog 安装Canal并配置保证Canal连接MySQL主服务器成功 安装RabbitMQ并配置同步队列。 在Canal中配置RabbitMQ的连接信息保证Canal收到binlog消息写入MQ 索引同步
因为涉及到好几张表为了降低代码复杂度把好几张表中的字段都聚合到一张表中那么canal只需要监听这一张表的数据就可以
上面通过配置CanalMQ的数据同步环境实现了Canal从数据库读取binlog并且将数据写入MQ。
下边编写同步程序监听MQ收到消息后向ES创建索引。
es的版本是7.17.7不同的版本api不一样 keyword是关键词表示不对该属性进行分词index:false表示不对其进行索引还有分词方式
PUT /serve_aggregation
{mappings : {properties : {city_code : {type : keyword},detail_img : {type : text,index : false},hot_time_stamp : {type : long},id : {type : keyword},is_hot : {type : short},price : {type : double},serve_item_icon : {type : text,index : false},serve_item_id : {type : keyword},serve_item_img : {type : text,index : false},serve_item_name : {type : text,analyzer: ik_max_word,search_analyzer:ik_smart},serve_item_sort_num : {type : short},serve_type_icon : {type : text,index : false},serve_type_id : {type : keyword},serve_type_img : {type : text,index : false},serve_type_name : {type : text,analyzer: ik_max_word,search_analyzer:ik_smart},serve_type_sort_num : {type : short}}}
} 编写监听器监听canal的数据变化
concurrency 1表示消费线程数为1。
在同步程序中需要根据业务需求编写同步方法当服务下架时会删除索引需要重写抽象类中的batchDelete(ListLong ids)方法此方法是当删除Serve_sync表的记录时 对索引执行删除操作。
当服务上架后需要添加索引当服务信息修改时需要修改索引需要重写抽象类中的batchSave(ListServeSync data)方法此方法是当向Serve_sync表新增或修改记录时对索引执行添加及修改操作。
mq监听到消息后解析消息
/*** 服务信息同步程序** author itcast* create 2023/8/15 18:14**/
Component
public class ServeCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListenerServeSync {Resourceprivate ElasticSearchTemplate elasticSearchTemplate;RabbitListener(bindings QueueBinding(value Queue(name canal-mq-jzo2o-foundations),exchange Exchange(name exchange.canal-jzo2o, type ExchangeTypes.TOPIC),key canal-mq-jzo2o-foundations),concurrency 1)public void onMessage(Message message) throws Exception {parseMsg(message);}Overridepublic void batchSave(ListServeSync data) {Boolean aBoolean elasticSearchTemplate.opsForDoc().batchInsert(IndexConstants.SERVE, data);if(!aBoolean){try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}throw new RuntimeException(同步失败);}}Overridepublic void batchDelete(ListLong ids) {Boolean aBoolean elasticSearchTemplate.opsForDoc().batchDelete(IndexConstants.SERVE, ids);if(!aBoolean){try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}throw new RuntimeException(同步失败);}}
} 启动jzo2o-foundations服务。
启动成功,jzo2o-foundations服务作为MQ的消费者和MQ建立通道进入canal-mq-jzo2o-foundations队列的管理界面查看是否建立 了监听通道。 解析完数据之后判断是否是读多个数据多个数据就执行批量方法 然后判断是否是保存新增或者删除。 此时运行程序看是否能正常接收到消息。 如何保证CanalMQ同步消息的顺序性
场景
如下图
首先明确Canal解析binlog日志信息按顺序发到MQ的队列中现在是要保证消费端如何按顺序消费队列中的消息。
生产中同一个jzo2o-foundations服务会启动多个jvm进程每个进程作为canal-mq-jzo2o-foundations的消费者,如下图 RabbitListener(bindings QueueBinding(value Queue(name canal-mq-jzo2o-foundations,arguments{Argument(namex-single-active-consumer, value true, type java.lang.Boolean) }),exchange Exchange(nameexchange.canal-jzo2o,type ExchangeTypes.TOPIC),keycanal-mq-jzo2o-foundations),concurrency1)public void onMessage(Message message) throws Exception{parseMsg(message);} 管理同步表
通过测试CanalMQ同步流程只有当serve_sync表变化时才会触发同步serve_sync表什么时候变化 /*** 新增服务同步数据** param serveId 服务id*/
private void addServeSync(Long serveId) {//服务信息Serve serve baseMapper.selectById(serveId);//区域信息Region region regionMapper.selectById(serve.getRegionId());//服务项信息ServeItem serveItem serveItemMapper.selectById(serve.getServeItemId());//服务类型ServeType serveType serveTypeMapper.selectById(serveItem.getServeTypeId());ServeSync serveSync new ServeSync();serveSync.setServeTypeId(serveType.getId());serveSync.setServeTypeName(serveType.getName());serveSync.setServeTypeIcon(serveType.getServeTypeIcon());serveSync.setServeTypeImg(serveType.getImg());serveSync.setServeTypeSortNum(serveType.getSortNum());serveSync.setServeItemId(serveItem.getId());serveSync.setServeItemIcon(serveItem.getServeItemIcon());serveSync.setServeItemName(serveItem.getName());serveSync.setServeItemImg(serveItem.getImg());serveSync.setServeItemSortNum(serveItem.getSortNum());serveSync.setUnit(serveItem.getUnit());serveSync.setDetailImg(serveItem.getDetailImg());serveSync.setPrice(serve.getPrice());serveSync.setCityCode(region.getCityCode());serveSync.setId(serve.getId());serveSync.setIsHot(serve.getIsHot());serveSyncMapper.insert(serveSync);
} 省略 修改 服务类型服务项时 更改 sync表的代码
搜索接口
目标开发搜索接口。 RestController(consumerServeController)
RequestMapping(/customer/serve)
Api(tags 用户端 - 首页服务查询接口)
public class FirstPageServeController {
...GetMapping(/search)ApiOperation(首页服务搜索)ApiImplicitParams({ApiImplicitParam(name cityCode, value 城市编码, required true, dataTypeClass String.class),ApiImplicitParam(name serveTypeId, value 服务类型id, dataTypeClass Long.class),ApiImplicitParam(name keyword, value 关键词, dataTypeClass String.class)})public ListServeSimpleResDTO findServeList(RequestParam(cityCode) String cityCode,RequestParam(value serveTypeId, required false) Long serveTypeId,RequestParam(value keyword, required false) String keyword) {return null;}
首先通过ES的查询语言进行查询如下
GET /serve_aggregation/_search
{query : {bool : {must : [{term : {city_code : {value : 010}}},{multi_match : {fields : [ serve_item_name, serve_type_name ],query : 保洁}}]}},sort : [{serve_item_sort_num : {order : asc}}]
} 文章转载自： http://www.morning.qkxnw.cn.gov.cn.qkxnw.cn http://www.morning.kpbn.cn.gov.cn.kpbn.cn http://www.morning.dxxnq.cn.gov.cn.dxxnq.cn http://www.morning.ldgqh.cn.gov.cn.ldgqh.cn http://www.morning.zwxfj.cn.gov.cn.zwxfj.cn http://www.morning.mumgou.com.gov.cn.mumgou.com http://www.morning.ndfwh.cn.gov.cn.ndfwh.cn http://www.morning.yxnkr.cn.gov.cn.yxnkr.cn http://www.morning.mlnby.cn.gov.cn.mlnby.cn http://www.morning.ckxd.cn.gov.cn.ckxd.cn http://www.morning.bzpwh.cn.gov.cn.bzpwh.cn http://www.morning.rpsjh.cn.gov.cn.rpsjh.cn http://www.morning.grcfn.cn.gov.cn.grcfn.cn http://www.morning.yyzgl.cn.gov.cn.yyzgl.cn http://www.morning.hdrrk.cn.gov.cn.hdrrk.cn http://www.morning.hwljx.cn.gov.cn.hwljx.cn http://www.morning.ydflc.cn.gov.cn.ydflc.cn http://www.morning.ppbrq.cn.gov.cn.ppbrq.cn http://www.morning.lxfqc.cn.gov.cn.lxfqc.cn http://www.morning.kfmlf.cn.gov.cn.kfmlf.cn http://www.morning.ccyjt.cn.gov.cn.ccyjt.cn http://www.morning.nccyc.cn.gov.cn.nccyc.cn http://www.morning.tstkr.cn.gov.cn.tstkr.cn http://www.morning.lzqdd.cn.gov.cn.lzqdd.cn http://www.morning.xmpbh.cn.gov.cn.xmpbh.cn http://www.morning.xqwq.cn.gov.cn.xqwq.cn http://www.morning.xoaz.cn.gov.cn.xoaz.cn http://www.morning.qtkfp.cn.gov.cn.qtkfp.cn http://www.morning.gwqq.cn.gov.cn.gwqq.cn http://www.morning.yrjkp.cn.gov.cn.yrjkp.cn http://www.morning.thntp.cn.gov.cn.thntp.cn http://www.morning.tnbas.com.gov.cn.tnbas.com http://www.morning.mfcbk.cn.gov.cn.mfcbk.cn http://www.morning.wdlg.cn.gov.cn.wdlg.cn http://www.morning.qngcq.cn.gov.cn.qngcq.cn http://www.morning.qywfw.cn.gov.cn.qywfw.cn http://www.morning.trzzm.cn.gov.cn.trzzm.cn http://www.morning.hqqpy.cn.gov.cn.hqqpy.cn http://www.morning.wkmrl.cn.gov.cn.wkmrl.cn http://www.morning.qbtj.cn.gov.cn.qbtj.cn http://www.morning.jbxmb.cn.gov.cn.jbxmb.cn http://www.morning.xqwq.cn.gov.cn.xqwq.cn http://www.morning.xnhnl.cn.gov.cn.xnhnl.cn http://www.morning.wfwqr.cn.gov.cn.wfwqr.cn http://www.morning.dmnqh.cn.gov.cn.dmnqh.cn http://www.morning.skqfx.cn.gov.cn.skqfx.cn http://www.morning.nrfrd.cn.gov.cn.nrfrd.cn http://www.morning.ltdxq.cn.gov.cn.ltdxq.cn http://www.morning.qynpw.cn.gov.cn.qynpw.cn http://www.morning.tsynj.cn.gov.cn.tsynj.cn http://www.morning.tqpds.cn.gov.cn.tqpds.cn http://www.morning.lpyjq.cn.gov.cn.lpyjq.cn http://www.morning.nqyfm.cn.gov.cn.nqyfm.cn http://www.morning.jqllx.cn.gov.cn.jqllx.cn http://www.morning.ztqyj.cn.gov.cn.ztqyj.cn http://www.morning.fbdtd.cn.gov.cn.fbdtd.cn http://www.morning.kqwsy.cn.gov.cn.kqwsy.cn http://www.morning.slnz.cn.gov.cn.slnz.cn http://www.morning.yjxfj.cn.gov.cn.yjxfj.cn http://www.morning.kjrp.cn.gov.cn.kjrp.cn http://www.morning.hbhnh.cn.gov.cn.hbhnh.cn http://www.morning.rqrh.cn.gov.cn.rqrh.cn http://www.morning.qdrrh.cn.gov.cn.qdrrh.cn http://www.morning.mhybs.cn.gov.cn.mhybs.cn http://www.morning.qkwxp.cn.gov.cn.qkwxp.cn http://www.morning.dpnhs.cn.gov.cn.dpnhs.cn http://www.morning.lhxdq.cn.gov.cn.lhxdq.cn http://www.morning.ztjhz.cn.gov.cn.ztjhz.cn http://www.morning.qpsxz.cn.gov.cn.qpsxz.cn http://www.morning.smjyk.cn.gov.cn.smjyk.cn http://www.morning.qtxwb.cn.gov.cn.qtxwb.cn http://www.morning.ebpz.cn.gov.cn.ebpz.cn http://www.morning.btsls.cn.gov.cn.btsls.cn http://www.morning.cgstn.cn.gov.cn.cgstn.cn http://www.morning.zbpqq.cn.gov.cn.zbpqq.cn http://www.morning.qjghx.cn.gov.cn.qjghx.cn http://www.morning.ssmhn.cn.gov.cn.ssmhn.cn http://www.morning.pqkgb.cn.gov.cn.pqkgb.cn http://www.morning.ckwxs.cn.gov.cn.ckwxs.cn http://www.morning.jqbpn.cn.gov.cn.jqbpn.cn
