基于canal数据加工系统

客官°小女子只卖身不卖艺 2022-06-05 02:13 208阅读 0赞

# 数据加工系统开发文档 #

## 1.数据加工系统主流程 ##

![这里写图片描述][SouthEast]

## 2.服务介绍 ##

整个框架由4个服务组成：canal-server、canal-client、bimq-consumer、crm-bi-task四个服务组成

**canal-server**:主要负责获取mysql服务器的binlog日志，按照用户提交的事务维度解析为结构化数据，然后存储于内存中，并提供tcp服务。

**canal-client**:通过connect到canal-server从内存中pull数据，以mq.schedule.table.queue的方式建立消息队列，发送具体的消息到队列，消息按照每条数据库记录变更前、变更后、操作类型（insert、update、delete）形式组织。

**bimq-consumer**：主要负责从具体的数据库记录变更到数据加工系统数据的提取过程，其一般通过监控一个表或者几个表的数据，通过业务实现具体的业务逻辑（如多个表格的join，或者一些magic number到枚举类型的转化过程）获取到最终可以使用的数据加工系统数据。其中根据具体业务场景的不同提供了三种方式将数据加工系统存储到elasticsearch中，从而提供给具体业务分析和使用该数据。  
方式1（实时）：该方式一般是用于处理近实时和增量的用户数据的新增、更新等操作。  
方式2（历史）：用户也可以通过自己实现定时任务，从数据库中提取最终的数据加工系统数据，然后更新到elasticsearch中。  
方式3（第三方）：除了支持基于该框架来实现数据加工系统数据的更新外，还可以通过调用提供的dubbo服务来更新数据，从而解决一些数据不是存储到mysql的情形。  
另外该服务还提供了通过sql来查询elasticsearch中数据加工系统数据的操作，例如支持基本查询语句（select）、聚合操作(group)、度量操作（sum、count、avg)。

**crm-bi-task**：主要负责将用户提供的将数据加工系统数据写到elasticsearch中的功能。  
综合来说：业务方只需要在bimq-consumer中完成从mysql变更数据到数据加工系统数据的业务逻辑开发和历史数据提取的工作。

## 3.开发流程 ##

### 3.1 申请数据库表的日志变更监控（业务方完成） ###

业务方如果希望实现通过监控数据表的方式来实时更新数据加工系统数据，需要将要监控的数据库和表名给运维人员或者相关负责人。  
如：highso\_db1.callthinklog

### 3.2 配置数据表的监控（运维人员完成） ###

服务名：canal(即canal-server)  
canal是支持不重启服务，新增数据库表监控的。相关运维人员只需要按照格式，新增具体的配置项到服务配置目录下就行了，canal会每隔5秒扫描指定目录下的配置，如果有配置的变更，会启动或者重启相关的instance，从而达到不需要重新启动jvm而新增新的监控的目的，同时在新增监控时，也不会影响现有服务的使用。  
新增配置如下：

1）在canal.property中修改属性：canal.destinations 如：添加callthinlog canal.destinations=looyudata 到canal.destinations=looyudata,callthinklog  
2）copy已经有的文件夹重名为callthinklog，然后更改canal.instance.defaultDatabaseName和canal.instance.filter.regex（如果数据库的连接信息有变更需要，更新相关连接信息）。

当更新完配置后，最迟5秒后该表的canal-server就已经启动成功

### 3.3 启动client获取server中的数据信息（运维人员完成） ###

服务名：bimq-connector-canal（即canal-client）  
启动client：更改client【bimq-connector-canal】在disconf上的配置即可自动启动client。如：canal.destinations=xxx.table1 为 canal.destinations=xxx.table1,xxx.table2。  
程序会根据改属性的变更，自动的启动或者关闭与server连接的client服务。  
另外提供了CanalOperateController来实现单个client的手动启动、关闭和获取所有正在运行的client列表

### 3.4 业务开发数据加工系统逻辑（业务方） ###

服务名：bimq-connector-consumer

#### 3.4.1 数据加工系统mq-consumer开发【实时】 ####

对于增量或者实时的数据加工系统数据的新增或者更新，只需要实现一个mq的consumer来监听由canal-client产生的mq消息即可。  
1. 业务方需要定义一个数据加工系统的DTO，并继承于BiSerializable，并放在同一个目录下。【因为代码会自动扫描该目录下的类并建立相关的数据加工系统需要的mq队列等信息，如CustomerDTO 对应queue=mq.T1DTO.queue routing=mq.T1DTO.routing

/** * 用户TDO */
    @Data
    public class T1DTO extends BiSerializable { 
    
        /** * 用户ID */
        private Long id;
    
        /** * 注册时间 */
        private Date registerdate;
    
        /** * 注册位置 */
        private String registerplace;
    
    }

1.  实现一个mq消费者并集成，用于消费canal-client产生的机会，具体代码可以参见T1Consumer。
    
    定义要消费的queue的名字常量：mq.数据库名.数据表名.queue

public class QueueConsumerConstants { 
        public static final String T1_QUEUE = "canal.xxxx.t1.queue";
        public static final String T2_QUEUE = "canal.xxxx.t2.queue";
    
        public static final String T3_QUEUE = "canal.xxxx.t3.queue";
        public static final String T4_QUEUE = "canal.xxxx.t4.queue";
    
    
        public static final String T5_QUEUE = "canal.xxxx.t5.queue";
        public static final String T6_QUEUE = "canal.xxxx.t6.queue";
        public static final String T7_QUEUE = "canal.xxxx.t7.queue";
        public static final String T8_QUEUE = "canal.xxxx.t8.queue";
    }

继承AbstractConsumer,实现具体的业务方代码逻辑
    1.handleUpdate方法：beforeMap数据记录变更前的key-value  afterMap数据记录变更后的key-value. 当提取出需要数据加工系统DTO对象后，通过调用ConsumerMessageBuilder.buildUpdateMessageVo产生需要的返回值
    2.handleInsert方法：处理数据库的插入操作
    3.handleDelete方法：处理数据库的删除操作

@Component
    @RabbitListener(queues = QueueConsumerConstants.T1_QUEUE,
            containerFactory = "rabbitListenerContainerFactory")
    @Slf4j
    public class T1Consumer extends AbstractConsumer { 
    
        @Override
        protected List<ConsumerMessageBuilder.MessageVo> handleUpdate(String tableName,
                Map<String, Object> beforeMap, Map<String, Object> afterMap) throws Exception {
            return null;
        }
    
        @Override
        protected List<ConsumerMessageBuilder.MessageVo> handleInsert(String tableName,
                Map<String, Object> afterMap) throws Exception {
    
    
            Customer after = BeanCovertUtil.convertToJavaBeanIgnoreCase(afterMap, Customer.class);
            CustomerDTO dto = BeanCovertUtil.copy(after, CustomerDTO.class);
    
            dto.setNewKey(dto.getId().toString());
    
            log.info("新增用户,id:{}",after.getId());
    
            return Lists.newArrayList(ConsumerMessageBuilder.buildInsertMessageVo(dto));
        }
    
        @Override
        protected List<ConsumerMessageBuilder.MessageVo> handleDelete(String tableName,
                Map<String, Object> beforeMap) throws Exception {
            return null;
        }
    
    }

**PS**: 每个DTO对象都需要设置newKey或者oldKey，es会将其作为主键更新或者删除其数据  
其中的注解中的queues为用户需要监听的数据表队列

#### 3.4.2 数据加工系统定时任务开发【历史数据】 ####

定时任务主要解决历史数据加工系统数据的提取到elasticsearch的过程  
其中对于大家常用的单表的数据表遍历提供了一个封装,业务方只需要继承AbstractScanJdbcJob，并设置泛型参数domain对象和DTO对象即可。参考SynT1Data2ESJobV2

该方式的单表数据遍历主要是通过数据库表中的id自增主键进行切分，每次取1000条，转化为domain对象，然后通过调用covertToDTO转化为DTO对象，然后通过mq发送数据加工系统信息。

public class T1Data2ESJobV2 extends AbstractScanJdbcJob<Customer, CustomerDTO> { 
        @Resource(name = "haixueJdbcTemplate")
        private JdbcTemplate jdbcTemplate;
    
        @Override
        protected JdbcTemplate getJdbcTemplate() {
            return jdbcTemplate;
        }
    }

/** * bimq-connector Created by caowenyi on 2017/11/10 . */
    @Slf4j
    public abstract class AbstractScanJdbcJob<T, DTO extends BiSerializable> implements SimpleJob {
        @Resource
        private MessageSender messageSender;
    
        protected Class<T> tClass;
        protected Class<DTO> dtoClass;
        private String idField;
    
        protected String tableName;
        private RateLimiter rateLimiter = RateLimiter.create(50);
    
        protected abstract JdbcTemplate getJdbcTemplate();
    
        protected DTO covertToDTO(T obj) {
            return BeanCovertUtil.copy(obj, dtoClass);
        }
    
    
        private Long getId(T obj) {
            try {
                Field idField = tClass.getDeclaredField(this.idField);
                idField.setAccessible(true);
                Object id = idField.get(obj);
                return (Long) id;
            } catch (NoSuchFieldException | IllegalAccessException e) {
                log.error(e.getMessage(), e);
            }
            return null;
        }
    
        protected long getStartId() {
            return 0L;
        }
    
        protected int getPageSize() {
            return 1000;
        }
    
        public AbstractScanJdbcJob(){
            this("id");
        }
    
        public AbstractScanJdbcJob(String idField) {
            this.tClass = (Class) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
            this.dtoClass = (Class) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[1];
            this.tableName = tClass.getSimpleName().toLowerCase();
            this.idField=idField;
        }
    
        public AbstractScanJdbcJob(String idField,String tableName) {
            this.tClass = (Class) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
            this.dtoClass = (Class) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[1];
            this.tableName = tableName;
            this.idField=idField;
    
        }
    
        @Override
        public void execute(ShardingContext shardingContext) {
            long startId = getStartId();
            int pageSize = getPageSize();
            JdbcTemplate jdbcTemplate = getJdbcTemplate();
            while (true) {
                String sql = String.format("select * from %s where %s>%d order by %s limit %d ", tableName, idField, startId, idField, pageSize);
                List<T> objectList = jdbcTemplate.query(sql, (resultSet, row) -> {
                    Map<String, Object> hm = new HashMap<>();
                    ResultSetMetaData rsmd = resultSet.getMetaData();
                    int count = rsmd.getColumnCount();
                    for (int i = 1; i <= count; i++) {
                        String key = rsmd.getColumnLabel(i);
                        Object value = resultSet.getObject(i);
                        hm.put(key, value);
                    }
                    try {
                        return BeanCovertUtil.convertToJavaBeanIgnoreCase(hm, tClass);
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                    return null;
                });
                //List<T> objectList = jdbcTemplate.queryForList(sql, tClass);
                if (CollectionUtils.isEmpty(objectList)) {
                    break;
                }
                for (T obj : objectList) {
                    DTO dtoObj = covertToDTO(obj);
                    Long id = getId(obj);
                    if (id == null) {
                        throw new IllegalArgumentException("id字段不存在请重写getNewKey方法");
                    }
                    dtoObj.setNewKey(String.valueOf(id));
                    ConsumerMessageBuilder.MessageVo messageVo = ConsumerMessageBuilder.buildInsertMessageVo(dtoObj);
                    try {
                        messageSender.sendMsg(messageVo);
                        rateLimiter.acquire(1);
                    } catch (IllegalAccessException | InvocationTargetException | IntrospectionException e) {
                        log.error(e.getMessage(), e);
                    }
                    startId = id;
                }
            }
        }
    }

其中提供了如下几个protected方法用于兼容一些用户的细微不同：  
1. 用户需要提供具体使用的jdbc对象，因为不同的对象对应不同的jdbc对象名。具体参看：XXXX1DataSourceConfig、XXXX2DataSourceConfig、XXXX3DataSourceConfig

protected abstract JdbcTemplate getJdbcTemplate();

1.  支持用户转入的字段进行分片，在构造参数中传入改字段名即可

public AbstractScanJdbcJob(String idField) {
            this.tClass = (Class) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
            this.dtoClass = (Class) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[1];
            this.tableName = tClass.getSimpleName().toLowerCase();
            this.idField=idField;
    
        }

1.  支持用户设置tableName，默认为domain的名字
2.  支持用户设置开始数据遍历的起始位置和每次从数据库中获取的数量

protected long getStartId() {
            return 0L;
    }
    
    protected int getPageSize() {
            return 1000;
    }

PS：业务方也可以不用继承AbstractScanJdbcJob类，只需要MessageSender messageSender发送数据加工系统DTO数据即可（该方法有必传字段的限制）。

#### 3.4.3 其他方式的数据加工系统录入 ####

为支持一些非数据库方式的数据加工系统数据需要放到elasticsearch的需求，提供了RPC方式录入数据加工系统数据。

1.  引入依赖包

<dependency>
        <groupId>com.haixue</groupId>
            <artifactId>bimq-connector-remote</artifactId>
            <version>1.0.2-SNAPSHOT</version>
    </dependency>

1.  导入配置文件，并设置zk.hosts和dubbo.port两个值  
    springboot形式

@Configuration
    @ImportResource(locations = {
       "classpath*:applicationPersonal-client.xml"})
    public class DubboClientConfig { 
    
    }

spring形式  
在已经的xml bean配置文件中添加

代码：

@Slf4j
    public class RemoteSqlTest extends BaseTest{ 
        @Resource
        private MessageService biMessageService;
    
        @Test
        public void testMessage() throws Exception {
            CustomerDTO customerDTO=new CustomerDTO();
            customerDTO.setId(7749763L);
            customerDTO.setRegisterdate(DateUtils.parse("2017-11-08 15:52:15",PATTERN_YYYY_MM_DD_HH_MM_SS));
            customerDTO.setRegisterplace("4");
            customerDTO.setNewKey("7749763");
            ApiResponse apiResponse=biMessageService.sendInsertMessage(customerDTO);
            log.info(apiResponse.toString());
        }
    }

#### 3.4.4 数据加工系统数据的sql查询服务 ####

支持了基于sql的elasticsearch查询的dubbo服务，其jar包引入和spring配置参考 “3.4.3 其他方式的数据加工系统录入”

代码：

@Slf4j
    public class RemoteSqlTest extends BaseTest{ 
        @Resource
        private SqlSearchPersonalService sqlSearchPersonalService;
        @Test
        public void testSql() throws Exception {
            ApiResponse apiResponse=sqlSearchPersonalService.searchBySql("select * from crmcommunicationdto_index");
            log.info(apiResponse.toString());
        }
    }

### 3.5 crm-bi-task录入数据加工系统数据【bi开发人员】 ###

crm-bi-task服务通过实现数据加工系统队列的消费者来将业务方提供的数据加工系统数据录入到elasticsearch中。

1.建立es中的索引对象  
customerdto\_index

{
      "settings": { "index": { "number_of_replicas": "1", "number_of_shards": "6" } },
      "_default_": { "_all": { "enabled": false } },
      "mappings": { "customerdto_dim": { "date_detection": true, "dynamic_date_formats": [ "yyyy-MM-dd HH:mm:ss" ], "dynamic": "true", "dynamic_templates": [ { "strings": { "match_mapping_type": "string", "mapping": { "type": "string", "index": "not_analyzed" } } }, { "texts": { "match_mapping_type": "text", "mapping": { "type": "string", "index": "not_analyzed" } } }, { "dates": { "match_mapping_type": "date", "mapping": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd" } } } ], "properties": { "newKey": { "type": "string", "index": "not_analyzed" }, "oldKey": { "type": "string", "index": "not_analyzed" }, "id": { "type": "long" }, "registerdate": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd" }, "registerplace": { "type": "string", "index": "not_analyzed" } } } } }

1.  实现需要监听消息队列名常量

public class QueueConstants { 
        public static final String T1_QUEUE = "mq.T1DTO.queue";
        public static final String T2_QUEUE = "mq.T2DTO.queue";
        public static final String T3_QUEUE = "mq.T3DTO.queue";
        public static final String T4_QUEUE = "mq.T4DTO.queue";
        public static final String T5_QUEUE = "mq.T5DTO.queue";
        public static final String T6_QUEUE = "mq.T6DTO.queue";
        public static final String T7_QUEUE = "mq.T7DTO.queue";
        public static final String T8_QUEUE = "mq.T8DTO.queue";
    
    }

1.  实现消费者  
    只需要继承AbstractConsumer，然后传入需要录入的DTO对象即可。

@Component
    @RabbitListener(queues = QueueConstants.CUSTOMER_QUEUE, containerFactory = "rabbitListenerContainerFactory")
    public class T1Consumer extends AbstractConsumer<T1DTO> { 
    }

## 4.其他支持 ##

该系统除了支持数据加工系统的需求外，对于另外一些需要通过把数据库表同步到其他数据存储中间的清洗，例如solar等。业务方只需要申请要监控表格，然后监听具体的消息队列，然后处理数据库记录变更，并同步到其他中间件存储系统中。特别适合一些缓存的设置。

## 性能测试 ##

测试环境：  
1）canal-server/canal-client/bimq-connector-consumer都被部署在MacBook Pro单机环境中(处理器=2.2 GHz Intel Core i7 内存=16 GB 1600 MHz DDR3 图像卡=Intel Iris Pro 1536 MB 磁盘=500GB 操作系统=macOS Sierra JVM内存使用情况：-Xmx1g -Xms1g -Xmn512m  
2）mysql和rabbitmq都是单机部署在同一台虚拟机：虚拟机配置（处理器=Intel® Core™ i7-4770HQ CPU @ 2.20GHz、内存=992.3 MiB 图像卡=Gallium 0.4 on llvmpipe (LLVM 3.8, 256 bits) 磁盘=51.7 GB 操作系统类型=64位  
3) 单机测试由于性能限制（最大上限600qps)

具体性能如下：  
![这里写图片描述][SouthEast 1]  
4）总结延迟和数据操作的类型、当个操作影响的记录条数没有直接关系，延迟维持在350ms，整个服务的性能只和数据库、canal服务的部署环境有关，并且能够支持600qps以上的操作。

[SouthEast]: /images/20220605/d75037269a484cf1aac2792508d2d7d4.png
[SouthEast 1]: /images/20220605/3d8f10f4272d4420ac35de52796215af.png