需求:如上图这个查询,不须要太大的数据量就足以上MySQL数据库崩溃,解决方案之一就是将查询的数据同步到es服务器中,数据库持久层采用Mybatis + Mybatis-plusjava
gitee地址:https://gitee.com/ichiva/mybatis-sync-esgit
解决思路
使用Mybatis拦截器拦截update操做,获取拦截的类名+方法名组成的id,以及传入的参数,找到对应的处理器,交给处理器具体完成同步任务。处理器定义成接口,经过regHandler()方法注册进拦截器。key就是类名+方法名spring
核心代码
同步处理器接口
package com.gzwl.interceptor; /** * es 同步处理器 */ public interface SynEsHandler { void handler(Object parameter); }
拦截器
package com.gzwl.interceptor; import lombok.extern.slf4j.Slf4j; import org.apache.ibatis.executor.Executor; import org.apache.ibatis.mapping.MappedStatement; import org.apache.ibatis.plugin.*; import org.springframework.stereotype.Component; import java.util.*; /** * es 同步拦截器 * * 配置拦截器 @Signature * 前置拦截器 type = Executor.class * 只有写入才须要拦截 method = "update" * 拦截方法的参数 args = {MappedStatement.class, Object.class} */ @Slf4j @Component @Intercepts({ @Signature(type = Executor.class, method = "update", args = {MappedStatement.class, Object.class})}) public class SyncEsInterceptor implements Interceptor { private Map<string,syneshandler> handlerMap = new HashMap<>(); /** * 拦截器在sql执行成功后同步到es, * 若是同步失败抛出异常,保证数据一致性 * * @param invocation * @return * @throws Throwable */ @Override public Object intercept(Invocation invocation) throws Throwable { Object res = invocation.proceed(); Object[] args = invocation.getArgs(); if (args.length >= 2) { MappedStatement mappedStatement = (MappedStatement) args[0]; //拦截到的方法,也就是注册的key String key = mappedStatement.getId(); log.debug("拦截到的方法 key=",key); SynEsHandler synEsHandler = handlerMap.get(key); if(null != synEsHandler){ try { synEsHandler.handler(args[1]); }catch (Exception e){ //包装异常 throw new SyncEsException(e); } }else { log.debug("没有处理的key={}",key); } } return res; } @Override public Object plugin(Object o) { if(o instanceof Executor) return Plugin.wrap(o, this); else return o; } @Override public void setProperties(Properties properties) { } /** * 注册同步处理器 * @param key * @param parameterHandler */ public void regHandler(String key,SynEsHandler parameterHandler){ handlerMap.put(key,parameterHandler); } }
实现处理接口
package com.gzwl.interceptor.impl; import com.gzwl.entity.Employee; import com.gzwl.interceptor.SynEsHandler; import com.gzwl.interceptor.SyncEsInterceptor; import org.apache.ibatis.binding.MapperMethod; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component class EmployeeMapperHandler implements SynEsHandler { @Autowired public EmployeeMapperHandler(SyncEsInterceptor syncEsInterceptor){ //将本身注册到拦截器中,每一个方法可用是独立的handler,也可用注册多个方法的处理器 syncEsInterceptor.regHandler("com.gzwl.dao.EmployeeMapper.insert",this); syncEsInterceptor.regHandler("com.gzwl.dao.EmployeeMapper.update",this); } @Override public void handler(Object parameter) { Employee entity; if(parameter instanceof Employee){ //insert 可用直接拦截到实体类 entity = (Employee) parameter; }else { //update 方法须要特殊处理 entity = (Employee) ((MapperMethod.ParamMap) parameter).get("param1"); } //在这里编写es的同步逻辑 } }
完整代码:https://gitee.com/ichiva/mybatis-sync-essql
注意:这个方案也是有致命缺陷的,就是在es执行完成后,程序还有可能让数据库回滚,从而形成数据不一致。 </string,syneshandler>数据库