datalink拦截器（interceptor）开发竖表转横表同步

小咪咪 2022-09-09 09:46 84阅读 0赞

### datalink拦截器（interceptor）开发 ###

*  一、拦截器（interceptor）介绍
 *  二、拦截器（interceptor）开发流程
 *   *  1、在本机上写代码
     *  2、配置拦截器
 *  三、源码参考

# 一、拦截器（interceptor）介绍 #

拦截器可以对Records进行特殊处理，满足少数特定功能的需求，如过滤记录、Rdbms-Writer的内置拦截器DdlEventInterceptor对ddl同步的一些限制等等，不要在拦截器里做重功能开发（如读写三方库等）

# 二、拦截器（interceptor）开发流程 #

## 1、在本机上写代码 ##

//该例子是用来演示一个竖转横的demo,生产环境不建议这么做
    package com.ucar.datalink.writer.rdbms.intercept;
    
    import com.ucar.datalink.contract.log.rdbms.EventColumn;
    import com.ucar.datalink.contract.log.rdbms.EventType;
    import com.ucar.datalink.contract.log.rdbms.RdbEventRecord;
    import com.ucar.datalink.domain.RecordMeta;
    import com.ucar.datalink.domain.media.MediaColumnMappingInfo;
    import com.ucar.datalink.domain.media.MediaMappingInfo;
    import com.ucar.datalink.worker.api.intercept.Interceptor;
    import com.ucar.datalink.worker.api.task.TaskWriterContext;
    import com.ucar.datalink.worker.api.util.dialect.DbDialectFactory;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.jdbc.core.BeanPropertyRowMapper;
    import org.springframework.jdbc.core.JdbcTemplate;
    import org.springframework.jdbc.core.RowMapper;
    import org.springframework.util.CollectionUtils;
    import org.springframework.util.StringUtils;
    
    import java.io.Serializable;
    import java.util.HashMap;
    import java.util.HashSet;
    import java.util.List;
    import java.util.Map;
    import java.util.Set;
    
    /** * Created by lubiao on 2017/7/21. */
    public class TestInterceptor implements Interceptor<RdbEventRecord> { 
    
        private static final Logger logger = LoggerFactory.getLogger(TestInterceptor.class);
    
        @Override
        public RdbEventRecord intercept(RdbEventRecord record, TaskWriterContext context) { 
            try { 
                MediaMappingInfo mappingInfo = RecordMeta.mediaMapping(record);
                Boolean insertOrUpdate =
                    record.getEventType().equals(EventType.INSERT) || record.getEventType().equals(EventType.UPDATE);
                if ((!StringUtils.isEmpty(mappingInfo.getJoinColumn())) && insertOrUpdate) { 
                    String joinColumn = mappingInfo.getJoinColumn();
                    String[] joinColumnArray = joinColumn.split(",");
                    logger.info("joinColumnArray:{}", joinColumn);
                    //获取数据源
                    JdbcTemplate jdbcTemplate =
                        DbDialectFactory.getDbDialect(RecordMeta.mediaMapping(record).getSourceMedia().getMediaSource())
                            .getJdbcTemplate();
                    String sql =
                        "SELECT attrType,attrVal,testId from t_dl_test_attr where testId="+record.getKeys().get(0).getColumnValue();
    
                    //查询竖表数据
                    List<Map<String, Object>> columnValList = jdbcTemplate.queryForList(sql);
    
                    Map<String, String> columnMap = new HashMap<>(2);
                    if (!CollectionUtils.isEmpty(columnValList)) { 
                        columnValList.stream().forEach(testAttr -> { 
                            columnMap.put(testAttr.get("attrType").toString(), testAttr.get("attrVal").toString());
                        });
                    }
                    List<EventColumn> eventColumns = record.getColumns();
                    List<MediaColumnMappingInfo> columnMappingInfos=   mappingInfo.getColumnMappings();
    
    
                    Set<String> set =new HashSet<>();
                    int index = eventColumns.get(eventColumns.size() - 1).getIndex();
                    //在原有字段的后面加上竖表的字段
                    for (String columnName : joinColumnArray) { 
                        EventColumn eventColumn = new EventColumn();
                        eventColumn.setIndex(++index);
                        eventColumn.setKey(false);
                        eventColumn.setNull(false);
                        eventColumn.setColumnName(columnName);
                        eventColumn.setColumnValue(columnMap.get(columnName));
                        eventColumn.setUpdate(true);
                        eventColumn.setColumnType(12);
                        eventColumns.add(eventColumn);
                        set.add(columnName);
                    }
                    RecordMeta.attachExtendColumns(record,set);
                }
                return record;
            } catch (Throwable e) { 
                logger.error("TestInterceptor exception {}", e.getMessage());
                e.printStackTrace();
                return null;
            }
        }
    }

## 2、配置拦截器 ##

2.1新增拦截器  
![新增拦截器][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bGx5beF_size_20_color_FFFFFF_t_70_g_se_x_16]  
2.2 在映射管理中设置拦截器  
![设置拦截器][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bGx5beF_size_20_color_FFFFFF_t_70_g_se_x_16 1]

# 三、源码参考 #

[拦截器例子][Link 1]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bGx5beF_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/3eb0df43b6ce466d964292b8472ced8e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5bGx5beF_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/312227b852ce4d26b8a212262b64e8b8.png
[Link 1]: https://gitee.com/shandian534/DataLink/blob/xtc/dl-worker/dl-worker-writer-rdbms/src/main/java/com/ucar/datalink/writer/rdbms/intercept/TestInterceptor.java