一、前言：
最近学习了消息队列，刚好最近有一个需求：mysql数据更新或删除时，需要同步更新到Elasticsearch。所以写一篇简书来记录一下解决过程。如果还没有了解过消息队列，可以先翻看我前几篇简书。
RabbitMQ入门之安装 RabbitMQ入门之六种模式介绍（一）
RabbitMQ入门之六种模式介绍（二） SpringBoot中使用RabbitMQ
源码地址：https://github.com/rjqamdaflf/synchronize
二、思路：
- 对于RabbitMQ，我选用Work queues模型，因为这个模型中队列的消息一旦消费，就会消失，消息是不会被重复消费的。
- 生产与消费消息在两个小项目中，因此发送消息类型选择JSONObject，接收到消息后再利用反射机制创建对应实体类对象。
三、代码实现：
生产者：
消息内容：包含增删改需要调用的DAO接口名称，实体类名，以及实体类对象具体内容转成json字符串
//获取消息内容
public static JSONObject getEsMessage(String JpaBeanName, Object object) {
JSONObject json = new JSONObject();
//传入DAO接口名称
json.put("JpaBeanName", JpaBeanName);
//获取类名，这个类名是给消费者通过反射去创建实例化对象
String[] split = object.getClass().toString().split("\\.|\\s+");
json.put("ClassName", split[split.length - 1]);
//将对象转成json字符串
json.put("Object", JSONObject.toJSONString(object));
return json;
}
发送消息：在需要对mysql进行增删改操作的函数中，增加发送消息操作，例如下面这个例子：convertAndSend()函数中，第一个参数为交换机名称，这里不需要，第二个参数为key，第三个为发送内容。
@Override
public int saveCourse(Node courseNode) {
//存入数据库，返回存入的数据
Node newNode = nodeJpaRepository.save(courseNode);
//发送消息到消息队列
rabbitTemplate.convertAndSend("", "save", JsonUtils.getEsMessage(NameUtils.NODESERVICE_ES_BEAN_NAME, Node.toEsNode(newNode)));
return newNode.getId();
}
将DAO接口名字封装成一个类
public class NameUtils {
//接口名需要和消费者的接口名一致
public final static String CHAPTERSERVICE_ES_BEAN_NAME = "esChapterJpa";
public final static String COMMENTSERVICE_ES_BEAN_NAME = "esCommentJpa";
public final static String COURSESERVICE_ES_BEAN_NAME = "esCourseJpa";
public final static String KGSERVICE_ES_BEAN_NAME = "esKgJpa";
public final static String NODESERVICE_ES_BEAN_NAME = "esNodeJpa";
public final static String SCHOOLSERVICE_ES_BEAN_NAME = "esSchoolJpa";
public final static String SOURCESERVICE_ES_BEAN_NAME = "esSourceJpa";
}
消费者：
消息消费函数
@Component
public class EsCustomer {
//队列名
public final static String SAVE = "Save";
public final static String DELETE = "Delete";
//类基础路径
public final static String BASE_ENTITY_PACKAGE = "com.demo.synchronize.es.esentity.";
@Resource
private BeanUtils beanUtils;
//当此队列有消息时，说明mysql表有进行增改操作，读取队列里面的json数据，通过beanName获取对应bean实例
@RabbitListener(queuesToDeclare = @Queue(EsCustomer.SAVE))
public void NodeSave(JSONObject message) {
System.out.println("接收到 save消息：" + message);
String className = BASE_ENTITY_PACKAGE + message.getString("ClassName");
//通过反射机制，得到实例化对象
Object object = ClassUtils.getEsClass(className, message.getString("Object"));
System.out.println("反射得到的object:" + object);
//获取对应的bean
ElasticsearchRepository elasticsearchRepository = (ElasticsearchRepository) beanUtils.getBeanByBeanName(message.getString("JpaBeanName"));
//System.out.println(elasticsearchRepository);
if (elasticsearchRepository != null && object != null) {
//进行保存或修改操作
elasticsearchRepository.save(object);
}
}
@RabbitListener(queuesToDeclare = @Queue(EsCustomer.DELETE))
public void NodeDelete(JSONObject message) {
System.out.println("接收到 delete消息：" + message);
//获取对应的bean
ElasticsearchRepository elasticsearchRepository = (ElasticsearchRepository) beanUtils.getBeanByBeanName(message.getString("JpaBeanName"));
Object object = message.get("Object");
if (elasticsearchRepository != null && object != null) {
//删除操作
elasticsearchRepository.deleteById(object);
}
}
}
利用反射机制创建实体类对象
public static Object getEsClass(String className, String object) {
Class beanClass = null;
Object bean;
try {
// 加载 beanClass
beanClass = Class.forName(className);
//创建实体类对象
bean = beanClass.newInstance();
} catch (Exception e) {
e.printStackTrace();
//如果出错直接返回null，消费者判断null直接将消费扔掉
return null;
}
JSONObject jsonObject = JSONObject.parseObject(object);
jsonObject.keySet().forEach(s -> {
try {
// 利用反射将 bean 相关字段访问权限设为可访问
Field declaredField = bean.getClass().getDeclaredField(s);
declaredField.setAccessible(true);
//获取对应属性值
Object value = jsonObject.get(s);
// 将属性值填充到相关字段中
declaredField.set(bean, value);
} catch (Exception e) {
//可能会有部分属性名不匹配
e.printStackTrace();
}
});
return bean;
}
获取bean的方法，如果找不到这个bean或者有其他异常，直接返回null，消费者丢弃这条消息
@Component
public class BeanUtils {
@Resource
private ApplicationContext applicationContext;
public Object getBeanByBeanName(String beanName) {
try {
return applicationContext.getBean(beanName);
} catch (Exception e) {
return null;
}
}
}
四、效果：
1. 操作Node表：
控制台输出
es可视化工具查看
控制台输出
es可视化工具查看
2.操作chapter表：
控制台输出
es可视化工具查看
