1.MQTT框架，长时间运行容易掉线，而且还重连不上 (Springboot+MQTT)？

原因：使用相同的clientID
方案：全局使用的clientID保证唯一性，可以采用UUID等方式
原因: 当前用户没有Topic权限
方案：如果配置有acl权限，则查询当前登录mqtt用户是否具有订阅对应topic的权限，无权时也会造成一直频繁断线重连
原因：在回调函数内进行业务处理遇到异常并没有捕获
方案：在可能出现异常的语句块，进行try-catch捕获

  1.     /**
  2.      * subscribe订阅后得到的消息会执行到这里
  3.      */
  4.     @Override
  5.     public void messageArrived(String topic, MqttMessage message) {
  6.         String msg = new String(message.getPayload());
  7.         try {
  8.             //此处可能因为收到的消息不合法，会造成JSON转化异常，若异常未捕获，会导致MQTT客户端掉线
  9.             JSONObject jsonObject = JSON.parseObject(msg);
  10.             String gwId = String.valueOf(jsonObject.get("gwId"));
  11.         } catch (JSONException e) {
  12.             log.error("JSON Format Parsing Exception : {}", msg);
  13.         }
  14.     }

jmqtt-master.zip
iot-mqtt-server-master.zip

2.es里的数据导入另一个es？
答：reindexm，es有tool的。

https://gitee.com/ichiva/mybatis-sync-es

核心代码

同步处理器接口

  1. package com.gzwl.interceptor;
  3. /**
  4.  * es 同步处理器
  5.  */
  6. public interface SynEsHandler {
  8.     void handler(Object parameter);
  9. }

拦截器

  1. package com.gzwl.interceptor;
  3. import lombok.extern.slf4j.Slf4j;
  4. import org.apache.ibatis.executor.Executor;
  5. import org.apache.ibatis.mapping.MappedStatement;
  6. import org.apache.ibatis.plugin.*;
  7. import org.springframework.stereotype.Component;
  9. import java.util.*;
  11. /**
  12.  * es 同步拦截器
  13.  *
  14.  * 配置拦截器 @Signature
  15.  * 前置拦截器 type = Executor.class
  16.  * 只有写入才需要拦截 method = "update"
  17.  * 拦截方法的参数 args = {MappedStatement.class, Object.class}
  18.  */
  19. @Slf4j
  20. @Component
  21. @Intercepts({
  22.         @Signature(type = Executor.class, method = "update", args = {MappedStatement.class, Object.class})})
  23. public class SyncEsInterceptor implements Interceptor {
  25.     private Map<String,SynEsHandler> handlerMap = new HashMap<>();
  27.     /**
  28.      * 拦截器在sql执行成功后同步到es，
  29.      * 如果同步失败抛出异常，保证数据一致性
  30.      *
  31.      * @param invocation
  32.      * @return
  33.      * @throws Throwable
  34.      */
  35.     @Override
  36.     public Object intercept(Invocation invocation) throws Throwable {
  37.         Object res = invocation.proceed();
  39.         Object[] args = invocation.getArgs();
  40.         if (args.length >= 2) {
  41.             MappedStatement mappedStatement = (MappedStatement) args[0];
  43.             //拦截到的方法，也就是注册的key
  44.             String key = mappedStatement.getId();
  45.             log.debug("拦截到的方法 key=",key);
  47.             SynEsHandler synEsHandler = handlerMap.get(key);
  48.             if(null != synEsHandler){
  49.                 try {
  50.                     synEsHandler.handler(args[1]);
  51.                 }catch (Exception e){
  52.                     //包装异常
  53.                     throw new SyncEsException(e);
  54.                 }
  55.             }else {
  56.                 log.debug("没有处理的key={}",key);
  57.             }
  58.         }
  60.         return res;
  61.     }
  63.     @Override
  64.     public Object plugin(Object o) {
  65.         if(o instanceof Executor) return Plugin.wrap(o, this);
  66.         else return o;
  67.     }
  69.     @Override
  70.     public void setProperties(Properties properties) {
  72.     }
  74.     /**
  75.      * 注册同步处理器
  76.      * @param key
  77.      * @param parameterHandler
  78.      */
  79.     public void regHandler(String key,SynEsHandler parameterHandler){
  80.         handlerMap.put(key,parameterHandler);
  81.     }
  82. }

实现处理接口

  1. package com.gzwl.interceptor.impl;
  3. import com.gzwl.entity.Employee;
  4. import com.gzwl.interceptor.SynEsHandler;
  5. import com.gzwl.interceptor.SyncEsInterceptor;
  6. import org.apache.ibatis.binding.MapperMethod;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Component;
  10. @Component
  11. class EmployeeMapperHandler implements SynEsHandler {
  13.     @Autowired
  14.     public EmployeeMapperHandler(SyncEsInterceptor syncEsInterceptor){
  15.         //将自己注册到拦截器中,每个方法可用是独立的handler，也可用注册多个方法的处理器
  16.         syncEsInterceptor.regHandler("com.gzwl.dao.EmployeeMapper.insert",this);
  17.         syncEsInterceptor.regHandler("com.gzwl.dao.EmployeeMapper.update",this);
  18.     }
  20.     @Override
  21.     public void handler(Object parameter) {
  22.         Employee entity;
  24.         if(parameter instanceof Employee){
  25.             //insert 可用直接拦截到实体类
  26.             entity = (Employee) parameter;
  27.         }else {
  28.             //update 方法需要特殊处理
  29.             entity = (Employee) ((MapperMethod.ParamMap) parameter).get("param1");
  30.         }
  32.         //在这里编写es的同步逻辑
  33.     }
  34. }

完整代码：https://gitee.com/ichiva/mybatis-sync-es
注意：这个方案也是有致命缺陷的，就是在es执行完成后，程序还有可能让数据库回滚，从而造成数据不一致。