Praktische Statistiken zum Auftragseingang (1)-JavaBase-php.cn

RocketMQ部分

如果用的aliyun的RocketMQ,配置代码部分直接可参考文档自建的RocketMQ也可参照简单的消费例子监控对应的TOPIC即可消费Canal发来的数据,格式如下:

{
    "data":[
        {
            //单个修改后表数据,如果同一时间有多个表变动会有多个该JSON对象        }
    ],
    "database":"监控的表所在数据库",
    "es":表变动时间,
    "id":canal生成的id,
    "isDdl":Boolean类型,表示是否DDL语句,
    "mysqlType":{
        表结构
    },
    "old":如果是修改类型会填充修改前的值,
    "pkNames":[
        该表的主键,如"id"
    ],
    "sql":"执行的SQL",
    "sqlType":{
        字段对应的sqlType,一般使用mysqlType即可
    },
    "table":"监控的表名",
    "ts":canal记录发送时间,
    "type":"表的修改类型,入INSERT,UPDATE,DELETE"
}

MQ消费代码主要用了反射,映射到对应的表

//这里的body就是Canal发来的数据
public Action process(String body) {
        boolean result = Boolean.FALSE;
        JSONObject data = JSONObject.parseObject(body);
        log.info("数据库操作日志记录：data:{}",data.toString());
        Class c = null;
        try {
            //这里监控了订单和收益表分别做订单统计和收益日报统计
            c = Class.forName(getClassName(data.getString("table")));
        } catch (ClassNotFoundException e) {
            log.error("error {}",e);
        }
        if (null != c) {
            JSONArray dataArray = data.getJSONArray("data");
            if (dataArray != null) {
                //把获取到的data部分转换为反射后的实体集合
                List list = dataArray.toJavaList(c);
                if (CollUtil.isNotEmpty(list)) {
                    //对修改和写入操作分别进行逻辑操作
                    String type = data.getString("type");
                    if ("UPDATE".equals(type)) {
                        result = uppHistory(list);
                    } else if ("INSERT".equals(type)) {
                        result = saveHistory(list);
                    }
                }
            }
        }
        return result ? Action.CommitMessage : Action.ReconsumeLater;
    }
    
    /**
     * @description: 获取反射ClassName
     * @author: chenyunxuan
     */
    private String getClassName(String tableName) {
        StringBuilder sb = new StringBuilder();
        //判断是哪张表的数据
        if (tableName.equals("t_user_income_detail")) {
            sb.append("cn.mc.core.model.order");
        } else if (tableName.equals("t_user_cash_out")) {
            sb.append("cn.mc.sync.model");
        }
        String className = StrUtil.toCamelCase(tableName).substring(1);
        return sb.append(".").append(className).toString();
    }
    
    /**
     * @description: 写入对应类型的统计表
     * @author: chenyunxuan
     */
    private <T> Boolean saveHistory(List<T> orderList) {
        boolean result = Boolean.FALSE;
        Object dataType = orderList.get(0);
        //用instanceof判断类型进入不同的逻辑处理代码
        if (dataType instanceof TUserIncomeDetail) {
            result = userOrderHistoryService.saveIncomeDaily(orderList);
        } else if (dataType instanceof UserCashOut) {
            result = userCashOutHistoryService.delSaveHistoryList(orderList);
        }
        return result;
    }

saveIncomeDaily伪代码

  public synchronized Boolean saveIncomeDaily(List orderList) {
    //循环收益明细记录
    .......
    //通过创建时间和用户id查询收益日报表中是否有当日数据
    if(不存在当日数据){
        //创建当日的收益日报表记录
        .....
    }
    //因为不存在当日记录也会立即写入当日的空数据,所以下面的流程都是走更新流程
    //更新当日数据
    .......
    return Boolean.TRUE;
    }

注:代码中应该多打一些日志,方便产生异常收益数据后的校对

后记

至此一个基于canal+RocketMQ的收益日报统计异构方案就完成了,下一篇会围绕本文提到的第二个问题减少聚合SQL的产生展开.敬请关注.

相关免费学习推荐：java基础教程

Praktische Statistiken zum Auftragseingang (1)

Einführung

Anforderungen

Rendering

Spezifische Bedürfnisse

Denken

Sichtbare Probleme

Implementierung

Technologie-Stack

Projektprozess

Startup

RocketMQ部分

后记

In Verbindung stehende Artikel