标签：实战 orderId 事务批量文件案例线上日志上传

本文是我从业多年开发生涯中针对线上业务的处理经验总结而来，这些业务或多或少相信大家都遇到过，因此在这里分享给大家，大家也可以看看是不是遇到过类似场景。本文大纲如下，

后台上传文件

线上后台项目有一个消息推送的功能，运营新建一条通知消息时，需要一起上传一列包含用户 id 的文件，来给文件中包含的指定用户推送系统消息。

如上功能描述看着很简单，但是实际上处理上传文件这一步是由讲究的，假如说后台上传文件太大，导致内存溢出，又或者读取文件太慢等其实都是一些隐性的问题。

对于技术侧想要做好这个功能，保证大用户量（比如达到百万级别）下，上传文件、发送消息功能都正常，其实是需要仔细思考的，我这里给出我的优化思路，

上传文件类型选择

通常情况下大部分用户都会使用 Excel 文件作为后台上传文件类型，但是相比 Excel 文件，还有一种更加推荐的文件格式，那就是 CSV 文件。

CSV 是一种纯文本格式，数据以文本形式存储，每行数据以逗号分隔，没有任何格式化。

因此 CSV 适用于简单、易读、导入和导出的场景，而且由于 CSV 文件只包含纯文本，因此文件大小通常比 Excel 文件小得多。

但是 CSV 文件针对复杂电子表格操作的支持就没 Excel 功能那么强大了，不过在这个只有一列的文件上传业务里够用了。

假如说上传文件中包含 100 万用户 id，那么这里使用 CSV 文件上传就有明显优势，占用内存更少，处理上传文件也更快。

消息推送状态保存

由于大批量数据插入是一个耗时操作（可能几秒也可能几分钟），所以需要保存批量插入是否成功的状态，在后台中还需要显现出这条消息推送状态是成功还是失败，方便运营人员回溯消息推送状态。

批量写入

针对这里上传大文件时的批量写入场景，这里提几个点大家注意一下就行，

rewriteBatchedStatements=true

MySQL 的 JDBC 连接的 url 中要加 rewriteBatchedStatements 参数，并保证 5.1.13 以上版本的驱动，才能实现高性能的批量插入。

MySQL JDBC 驱动在默认情况下会无视 executeBatch()语句，把我们期望批量执行的一组 sql 语句拆散，一条一条地发给 MySQL 数据库，批量插入实际上是单条插入，直接造成较低的性能。只有把 rewriteBatchedStatements 参数置为 true，驱动才会帮你批量执行 SQL。另外这个选项对 INSERT/UPDATE/DELETE 都有效。

是否启用事物功能

批量写入场景里要不要启用事物，其实很多人都有自己的看法，这里我给出启用于不启用的利弊，

启用事务：好处在于如批量插入过程中，异常情况可以保证原子性，但是性能比不开事务低，在特大数据量下会明显低一个档次
不启用事务：好处就是写入性能高，特大数据量写入性能提升明显，但是无法保证原子性

在本文提到的大文件上传批量写入的场景下，要是追求极致性能我推荐是不启用事务的。

假如在批量写入过程中发生网络波动或者数据库宕机，我们其实只需要重新新建一条通知消息，然后重新上传包含用户 id 的文件即可。

因为上一条通知消息因为批量插入步骤没有全部完成，所以推送状态是失败。后续等开发人员处理一下脏数据即可。

大事务

@Transactional 是 Spring 框架提供得事务注解，相信这是许多人都知道的，但是在一些高性能场景下，是不建议使用的，推荐通过编程式事务来手动控制事务提交或者回滚，减少事务影响范围，因而提升性能。

使用事务注解

如下是一段订单超时未支付回滚业务数据得代码，采用 @Transactional 事务注解

@Transactional(rollbackFor = Exception.class)
public void doUnPaidTask(Long orderId) {
    // 1. 查询订单是否存在
    Order order = orderService.getById(orderId);
    ，，，

    // 2. 更新订单为已取消状态
    order.setOrderStatus((byte) OrderStatusEnum.ORDER_CLOSED_BY_EXPIRED.getOrderStatus());
    orderService.updateById(order);
    ...
    // 3. 订单商品数量增加
    LambdaQueryWrapper<OrderItem> queryWrapper = Wrappers.lambdaQuery();
    queryWrapper.eq(OrderItem::getOrderId， orderId);
    List<OrderItem> orderItems = orderItemService.list(queryWrapper);
    for (OrderItem orderItem : orderItems) {
        Long goodsId = orderItem.getGoodsId();
        Integer goodsCount = orderItem.getGoodsCount();
        if (!goodsDao.addStock(goodsId， goodsCount)) {
            throw new BusinessException("秒杀商品货品库存增加失败");
        }
    }

    // 4. 返还用户优惠券
    couponService.releaseCoupon(orderId);
    log.info("---------------订单orderId:{}，未支付超时取消成功"， orderId);
}

可以看到上面订单回滚的代码逻辑有四个步骤，如下，

查询订单是否存在
更新订单为已取消状态
订单商品数量增加
返还用户优惠券

这里面有个问题，订单回滚方法里面其实只有 2、3、4 步骤是需要在一个事物里执行的，第 1 步其实可以放在事物外面来执行，以此缩小事物范围。

使用编程式事务

使用编程式事务对其优化后，代码如下，

@Resource
private PlatformTransactionManager platformTransactionManager;
@Resource
private TransactionDefinition transactionDefinition;

public void doUnPaidTask(Long orderId) {
    // 启用编程式事务
    // 1. 在开启事务钱查询订单是否存在
    Order order = orderService.getById(orderId);
    ...
    // 2. 开启事务
    TransactionStatus transaction = platformTransactionManager.getTransaction(transactionDefinition);
    try {
        // 3. 设置订单为已取消状态
        order.setOrderStatus((byte) OrderStatusEnum.ORDER_CLOSED_BY_EXPIRED.getOrderStatus());
        orderService.updateById(order);
        ...
        // 4. 商品货品数量增加
        LambdaQueryWrapper<OrderItem> queryWrapper = Wrappers.lambdaQuery();
        queryWrapper.eq(OrderItem::getOrderId， orderId);
        List<OrderItem> orderItems = orderItemService.list(queryWrapper);
        for (OrderItem orderItem : orderItems) {
            Long goodsId = orderItem.getGoodsId();
            Integer goodsCount = orderItem.getGoodsCount();
            if (!goodsDao.addStock(goodsId， goodsCount)) {
                throw new BusinessException("秒杀商品货品库存增加失败");
            }
        }

        // 5. 返还优惠券
        couponService.releaseCoupon(orderId);
        // 6. 所有更新操作完成后，提交事务
        platformTransactionManager.commit(transaction);
        log.info("---------------订单orderId:{}，未支付超时取消成功"， orderId);
    } catch (Exception e) {
        log.info("---------------订单orderId:{}，未支付超时取消失败"， orderId， e);
        // 7. 发生异常，回滚事务
        platformTransactionManager.rollback(transaction);
    }
}

可以看到采用编程式事务后，我们将查询逻辑排除在事务之外，这样也就减小了事物影响范围。

在极高性能优先的场景下，我们甚至可以考虑不使用事务，使用本地消息表 + 消息队列来实现最终一致性就行。

海量日志采集

公司线上有一个项目的客户端，采用 tcp 协议与后端的一个日志采集服务建立连接，用来上报客户端日志数据。

在业务高峰期下，会有同时成千上万个客户端建立连接，实时上报日志数据。

在上面的高峰期场景下，日志采集服务会有不小的压力，如果程序代码逻辑处理稍有不当，就会造成服务卡顿、CPU 占用过高、内存溢出等问题。

为了解决上面的大量连接实施上报数据的场景，日志采集服务决定使用 Netty 框架进行开发。

这里直接给出日志采集程序使用 Netty 后的一些优化点，

采集日志异步化

针对客户端连接上报日志的采集流程异步化处理有三个方案，给大家介绍一下，

普通版：采用阻塞队列 ArrayBlockingQueue 得生产者消费者模式，对上报的日志数据进行异步批量处理，在此场景下，通过生产者将数据缓存到内存队列中，然后再消费者中批量获取内存队列的日志数据保存入库，好处是简单易用，坏处是有内存溢出风险。
进阶版：采用 Disruptor 队列，也是一个基于内存的高性能生产者消费者队列，消费速度对比 ArrayBlockingQueue 有一个数量级以上得性能提升，附简介说明：https://www.jianshu.com/p/bad7b4b44e48。
终极版：也是公司日志采集程序最后采用的方案。采用 kfaka 消息队列中间件，先持久日志上报数据，然后慢慢消费。虽然引入第三方依赖会增加系统复杂度，但是 kfaka 在大数据场景表现实在是太优秀了，这一点也是值得。

采集日志压缩

对上报后的日志如果要再发送给其他服务，是需要进行压缩后再处理，这一步是为了避免消耗过多网络带宽。

在 Java 里通常是指序列化方式，Jdk 自带得序列化方式对比 Protobuf、fst、Hession 等在序列化速度和大小的表现上都没有优势，甚至可以用垃圾形容。

Java 常用的序列化框架有下面这些，

JDK 自带的序列化：性能较差，占用空间大，无法跨语言，好处是简单易用，通用性强。
JSON：常用的 JSON 库有 Jackson、Gson、Fastjson 等。性能较好，占用空间少，跨语言支持广泛，但是无法序列化复杂对象。
Protocol Buffers：由 Google 开源，基于 IDL 语言定义格式，编译器生成对象访问代码。性能高效占用空间小，但是需要提前定义 Schema。
Thrift：Facebook 开源，与 Protocol Buffers 类似。定制生态不如 PB 完善，但是支持多语言交互。
Avro：Hadoop 生态圈序列化框架，支持数据隔离与进化，动态读写，性能可靠性好，占用空间较小。但是使用复杂，通用性较差。
Hessian：一款开源的二进制远程通讯协议，使用简单方法提供了RMI功能，主要用于面向对象的消息通信。支持跨平台、多语言支持、使用简单，缺点是传递复杂对象性能会下降，不适合安全性高的应用。

如果兼容性要求不高可以选择 JSON，如果要求效率以及传输数据量越小越好则 PB/Thrift/Avro/Hessian 更合适。

数据落库选型

像日志这种大数据量落库，都是新增且无修改得场景建议使用 Clickhouse 进行存储，好处是相同数据量下对比 MySQL 占用存储更少，查询速度更快，坏处就是并发查询性能比较低，相比 MySQL 使用不算那么成熟。

最后聊两句

到这里本文所介绍三个线上业务优化实战就讲完了，其实这种实战案例还有很多，但是碍于篇幅本文就没讲那么多拉，后续有机会也会继续更新这类文章，希望大家能够喜欢。

关注公众号【waynblog】，每周分享技术干货、开源项目、实战经验、高效开发工具等，您的关注将是我的更新动力
标签：实战,orderId,事务,批量,文件,案例,线上,日志,上传
From： https://www.cnblogs.com/waynaqua/p/17893173.html

线上业务优化之案例实战