一、前言：

    最近学习了消息队列，刚好最近有一个需求：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;
        }
    }
}