上一期讲到了通过canal订阅mysql的binlog日志并且转换为对象，那么这一次我们将订阅来的对象通过RocketMQ发送消息，接收方接受消息之后同时存储到其他类型的数据源当中，完成一个简单的数据异构的过程。

什么是Java消息服务？

两个应用程序之间进行异步通信的API，它为标准消息协议和消息服务提供了一组通用接口，包括创建、发送、读取消息等，用于支持JAVA应用程序开发。

在J2EE中，当两个应用程序使用JMS进行通信时，它们之间并不是直接相连的，而是通过一个共同的消息收发服务连接起来，可以达到解耦的效果，我们将会在接下来的教程中详细介绍。

jms的消息传送模型

常见的消息传送模型有以下两种：

点对点消息传送模型

在点对点消息传送模型中，应用程序由消息队列，发送者，接收者组成。每一个消息发送给一个特殊的消息队列，该队列保存了所有发送给它的消息(除了被接收者消费掉的和过期的消息)。如下图所示：

发布订阅消息传送模型

在发布订阅模型中，消费者需要订阅相关的topic才能接收到生产者的信息。生产者会将信息传输到topic中，然后消费者只需要从topic中获取数据即可。如下图所示：

RocketMQ消息队列使用

这次使用的消息中间件为RocketMQ的使用场景。RocketMQ是阿里巴巴在2012年开源的分布式消息中间件，目前已经捐赠给Apache基金会，并于2016年11月成为 Apache 孵化项目。

RocketMQ在使用之前，需要我们引入相关的依赖配置：

    <!-- 整合RocketMq -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>${rocketmq.version}</version>
        </dependency>

关于RocketMQ的安装在这里就不做过多的讲解了。

通过mq的方式来进行数据异构通常是比较简单的方案，首先我们需要在项目里面独立一个模块专门用于监听mysql的binlog日志，这个模块我暂且称之为datahandle-core模块

整个工程采用了springboot的结构来构建，主要的核心也是在core工程中。

首先是监听canal的日志状态模块了，采用了上一节中讲解到的客户端代码进行数据监听，并且将其转换为对象然后发送往mq中：

package com.sise.datahandle.core;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

import static com.sise.datahandle.constants.CanalConstants.*;

/**
 * @author idea
 * @date 2019/10/20
 */
@Component
@Slf4j
public class CanalListener implements CommandLineRunner {

    @Autowired
    private CanalClient canalClient;

    @Override
    public void run(String... args) throws Exception {
      log.info("=============canal监听器开启===============");
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(SERVER_ADDRESS, PORT), DESTINATION, USERNAME, PASSWORD);
        canalConnector.connect();
        canalConnector.subscribe(".*\\..*");
        canalConnector.rollback();
        for (; ; ) {
            Message message = canalConnector.getWithoutAck(100);
            long batchId = message.getId();
            if (batchId != -1) {
                canalClient.entityHandle(message.getEntries());
            }
        }
    }
}

ps:这里面的CanalClient代码主要来自上一篇的canal客户端代码，文末会有完整项目代码链接，需要的读者可以前往查看。

在CanalClient里面，有一个函数是专门用于处理将订阅的数据发送到mq消息队列中：

package com.sise.datahandle.core;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import com.sise.datahandle.handler.CanalDataHandler;
import com.sise.datahandle.model.TypeDTO;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * canal监听客户端变化
 *
 * @author idea
 * @date 2019/10/12
 */
@Slf4j
@Service
public class CanalClient {

    @Autowired
    private DefaultMQProducer rocketMqProducer;

    /**
     * 处理binlog日志的监听
     *
     * @param entries
     */
    public void entityHandle(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }
            try {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    switch (rowChange.getEventType()) {
                        case INSERT:
                            String tableName = entry.getHeader().getTableName();
                            //测试选用t_type这张表进行映射处理
                            if ("t_type".equals(tableName)) {
                                TypeDTO typeDTO = CanalDataHandler.convertToBean(rowData.getAfterColumnsList(), TypeDTO.class);
                                org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message();
                                message.setTopic("canal-test-topic");
                                message.setTags("canal-test-tag");
                                String json = JSON.toJSONString(typeDTO);
                                message.setBody(json.getBytes());
                                SendResult sendResult = rocketMqProducer.send(message);
                                log.info("[mq消息发送结果]----" + sendResult);
                            }
                            break;
                        default:
                            break;
                    }
                }
            } catch (InvalidProtocolBufferException e) {
                log.error("[CanalClient]监听数据过程出现异常，异常信息为{}", e);
            } catch (InterruptedException | RemotingException | MQClientException | MQBrokerException e) {
                log.error("[CanalClient] mq发送信息出现异常:{}", e);
            }
        }
    }

}

这里面主要是监听binlog记录为插入数据事件的时候做发送mq操作。

接下来便是常见的mq配置了，本工程主要是一个模拟的简单案例，因此我将consumer和producer都放在了一起方便测试。

通过springboot自身的properties文件对mq进行参数初始化配置之后便可以构建一个基本的consumer和producer了。这里我们拿一个TypeDto类来进行树异构的测试，consumer端的核心代码为：

package com.sise.datahandle.mq.rocketmq.consumer;

import com.sise.datahandle.model.TypeDTO;
import com.sise.datahandle.mq.rocketmq.producer.RocketMqMsgHandle;
import com.sise.datahandle.redis.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;

/**
 * @author idea
 * @date 2019/10/20
 */
@Component
@Slf4j
public class RocketMqConsumeMsgListenerProcessor implements MessageListenerConcurrently {

    @Autowired
    private RedisService redisService;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        if(CollectionUtils.isEmpty(msgs)){
            log.info("接受到的消息为空，不处理，直接返回成功");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt messageExt = msgs.get(0);
        System.out.println("接受到的消息为："+messageExt.toString());
        if("canal-test-topic".equals(messageExt.getTopic())){
            if("canal-test-tag".equals(messageExt.getTags())){
                int reconsume = messageExt.getReconsumeTimes();
                if(reconsume ==3){//消息已经重试了3次，如果不需要再次消费，则返回成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                TypeDTO typeDTO = RocketMqMsgHandle.parseMessage(messageExt,TypeDTO.class);
                //存储进入redis中
                redisService.setObject("typeDTO-"+System.currentTimeMillis(),typeDTO);
            }
        }
        // 如果没有return success ，consumer会重新消费该消息，直到return success
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

通过订阅mq的信息，读取相关的数据再次写入到redis里面，完成一个简单过程的数据异构。

整个迷你工程写下来，比较核心的地方就在于对binlog日志的解析器部分，如何将日志订阅之后转换为相应的对象进行处理。

通常采用mq的方式进行数据异构会相对简单，实际上是在监听binlog为写DB的同时去写一次MQ，但是这种方式不能够保证数据一致性，就是不能保证跨资源的事务。注：调用第三方远程RPC的操作一定不要放到事务中。

完整案例的代码链接如下(点击阅读原文直达)：

https://gitee.com/IdeaHome_admin/wfw