주문수입의 편리한 통계 (1)

Basic Java Tutorial 컬럼에서는 주문 수입을 편리하게 계산하는 방법을 보여줍니다.

소개

통계적인 주문 수익은 전자상거래 앱에서 흔히 발생하는 문제입니다. 일반적인 요구 사항은 대략 사용자 수익입니다.日报/月报/年报.这些报表型的数据对表设计和程序设计有着不小的挑战.常规的聚合查询수익 테이블 데이터가 증가함에 따라 명세서 쿼리 시간은 점차 늘어납니다. 가변 길이가 커집니다. 이때 손익계산서를 어떻게 효율적으로 조회할 수 있는지 고민해야 합니다. 어떤 디자인이 통계적 소득을 단순화할 수 있을까요?

요구사항

렌더링

특별한 요구사항

• 소득 유형은 자체 구매 주문 소득, 공유 주문 소득, 유통수입, 활동수입
• 통계 당일수입, 이달수입
• 해당 기간의 수입은 필터링된 시간을 기준으로 계산됩니다.

Thinking

Design Ideas

주문표는 꼭 필요합니다. 주문표를 작성하거나 수정할 때 수정된 수입표도 동시에 작성됩니다. 주문은 주문표에 기록되며, 특별 사업의 손익계산서에만 기록됩니다. 그런 다음 사용자 收益日报表.单行记录写入用户当天收益情况.降低查询用户日/月/年收益统计时的数据量.以单用户为例,通过拆分用户一个月只会产生最多31 데이터를 생성하는 차원으로 사용합니다. 손익 계산서 데이터의 양은 사용자가 주문한 수와 일대일로 일치하므로 손익 계산서를 계속 사용하십시오. 사용자 수가 증가하기 시작하는 초기 단계에서는 대용량 데이터 통계를 피하기 위해 이 방법을 사용할 수 있습니다. 나중에 사용자 수가 증가하면 일일 보고서의 데이터가 커지면 하위를 고려할 수 있습니다. -테이블.

눈에 보이는 문제

• 일별 소득 보고서 동기화 타이밍은 원래 주문 작업이 매우 복잡하고 소득의 동기 작성과 일일 소득 데이터의 계산 및 작성이 필요하기 때문에 코드 결합이 손익계산서 이질성을 통해 일일 소득 보고서를 생성하는 방법은 무엇입니까?
• 일별 보고서에 소득이 기록되지만 렌더링에서 요구하는 효과를 충족하려면 여러 SQL 문이 필요할 수 있습니다. 결과에 영향을 주지 않고 이를 수행할 수 있는 방법이 있습니까? 프로그램 효율성을 유지하면서 집계된 SQL을 최대한 적게 사용하는 것은 어떻습니까?

구현 데이터 수집을 시작했고 마침내 이종 솔루션으로 채택되었습니다.

canal+RocketMQ

기술 스택이 두 가지 기술 프레임워크에 대한 간략한 소개:

• canal: 주요 목적은 MySQL 데이터베이스 증분 로그 분석을 기반으로 증분 데이터 구독 및 소비를 제공하는 것입니다.
• RocketMQ: 고가용성 분산 클러스터 기술을 기반으로 하는 오픈 소스 분산 메시징 시스템으로, 낮은 대기 시간 A를 제공합니다. 안정적인 메시지 게시 및 구독 서비스입니다.

注:我用的aliyun的全家桶,MQ和mysql都是阿里云的,如果是自建服务器的可能有区别,我在后面尽量标出

프로젝트 프로세스

1. 손익계산서를 작성하거나 수정하는 동안 운하를 통해 mysql 손익계산서의 binlog 로그를 모니터링하세요.
2. 운하는 변화를 감지하고 조립합니다. 변경된 JSON 메시지를 전송하고 RocketMQ에 사전 정의된 TOPIC을 보냅니다.

canal 구성 부분

공식 문서를 참조하세요. 압축을 풀면 세 개의 디렉터리가 포함된 운하 폴더가 생성됩니다

• bin: 시작 및 다시 시작 스크립트 저장
• conf:存放核心配置文件
• lib: 핵심 jar 패키지 저장

우리는 conf/canal.properties 핵심 구성 파일과 conf/example/에 집중해야 합니다. conf 폴더 instance.properties 단일 모니터링 노드 구성 파일

conf/canal.properties

# tcp, kafka, RocketMQ,这里默认是tcp读取模式,采用RocketMQ需要将其改变为RocketMQ模式
canal.serverMode = RocketMQ
# 如果是aliyun的RocketMQ需要配置以下两个KEY,ak/sk
canal.aliyun.accessKey =xxxxxxx
canal.aliyun.secretKey =xxxxxxx
# 监控的节点名称.这个默认就是example如果有多节点可以逗号隔开,如下方的例子
canal.destinations = example,sign
# 如果是aliyun的RocketMQ需要修改canal.mq.accessChannel为cloud默认为local
canal.mq.accessChannel = cloud
#MQ的地址,需要注意这里是不带http://,但是需要带端口号
canal.mq.servers = 
#rocketmq实例id
canal.mq.namespace =

conf/example/instance.properties

#mysql地址
canal.instance.master.address=
#以下两个参数需要在开启数据库binlog日志后得到,在数据库查询界面输入查询语句`show master status`,canal.instance.master.journal.name对应File参数,canal.instance.master.position对应Position参数
canal.instance.master.journal.name=
canal.instance.master.position=
#数据库的账号密码
canal.instance.dbUsername=
canal.instance.dbPassword=
#需要监控变动的表
canal.instance.filter.regex=xxx.t_user_order,xxx.t_user_cash_out
#定义发送的mq生产组
canal.mq.producerGroup = 
#定义发送到mq的指定主题
canal.mq.topic=

참고: 쓰기 규칙 형식의 경우 모니터링 테이블은 모니터링 테이블 작성 규칙을 참조하세요

Startup

cd /canal/bin
./start.sh

이때 canal 디렉토리에 추가 로그 파일을 찾아 들어가시면 운하 메인 로그 파일을 보실 수 있습니다. 예시 노드 시작 로그입니다.

canal日志中出现
 the canal server is running now ......
example日志中出现
 init table filter : ^tablename
 xxxxxxxxx , the next step is binlog dump

는 성공했다는 의미입니다. 큰 발전을 이루었으며 이제 운하 모니터링이 정상적으로 실행되고 있습니다.

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基础教程

위 내용은 주문수입의 편리한 통계 (1)의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!