博主這里的大資料量、高并發業務處理優化基于博主線上專案實踐以及全網資料整理而來，在這里分享給大家

一. 大資料量上傳寫入優化

線上業務后臺專案有一個訊息推送的功能，通過上傳包含用戶id的檔案，給指定用戶推送系統訊息

1.1 如上功能描述很簡單，但是對于技術側想要做好這個功能，保證大用戶量（比如達到百萬級別）下，系統正常運行，功能正常其實是需要仔細思考的，博主這里給出思路：

上傳檔案型別選擇

通常情況下大部分用戶都會使用excel檔案，但是相比excel檔案還有一種更加推薦的檔案格式，那就是csv檔案，相比excel檔案它可以直接在記事本編輯，excel也可以打開cvs檔案，且占用記憶體更少（畫重點），對于上傳的csv檔案過于龐大，也可以采用流式讀取，讀一部分寫一部分

訊息推送成功與否狀態保存

由于大批量資料插入是一個耗時操作（可能幾秒也可能幾分鐘），所以需要保存批量插入是否成功的狀態，在后臺中可以顯現出這條訊息推送記錄是成功還是失敗，方便運營回溯訊息推送狀態

批量寫入啟不啟用事務

博主這里給出兩種方案利弊：

啟用事務：好處在于如批量插入程序中，例外情況可以保證原子性，但是性能比不開事務低，在特大資料量下會明顯低一個檔次
不啟用事務：好處就是寫入性能高，特大資料量寫入性能提升明顯，但是無法保證原子性，但是對于已經批量插入的新增資料，只是會產生臟資料而已，在功能設計合理的情況下是不影響業務的，如下面第四點

綜上：在大資料量下，我們要是追求極致性能可以不啟用事務，具體選擇也需各位結合自身業務情況

推送例外失敗的訊息處理

建議功能設計上，可以屏蔽對失敗訊息再進行操作，這樣不需要再處理之前推送失敗寫入的臟資料，直接新增訊息推送即可

1.2 批量寫入代碼優化

jdbc引數攜帶 rewriteBatchedStatements=true 在jdbc驅動上啟動批量寫入功能，如下

spring.datasource.master.jdbc-url=jdbc:mysql://localhost:3306/test_db?allowMultiQueries=true&characterEncoding=utf8&autoReconnect=true&useSSL=false&rewriteBatchedStatements=true

啟用 insert into table(id, name) values(1, 'tom'),(2, 'jack') 模式，建議一次寫入個數不要太多，MySQL對于sql長度是有限制的，對于這種欄位少的表，一次寫入500 - 1000問題不大，欄位多了需要降低這個寫入量

insert into im_notice_app_ref(notice_id, app_id, create_time)
values
<foreach collection="list" separator="," item="item">
    (#{item.noticeId}, #{item.appId}, #{item.createTime})
</foreach>

一般情況下大家都知道第二條優化，但是可能會忽略jdbc引數攜帶 rewriteBatchedStatements=true，這個引數能在第二條的基礎上啟用批量執行SQL，進一步提升寫入性能

二. 大事務優化，減小影響范圍，提升系統處理能力

@Transactional 大于 Spring 提供得事務注解，許多人都知道，但是在高并發下，不建議使用，推薦通過編程式事務來手動控制事務提交或者回滾，減少事務影響范圍

如下是一段訂單超時未支付回滾業務資料得代碼，采用 @Transactional 事務注解

@Transactional(rollbackFor = Exception.class)
public void doUnPaidTask(Long orderId) {
    // 1. 查詢訂單是否存在
    Order order = orderService.getById(orderId);
    if (order == null) {
        throw new BusinessException(String.format("訂單不存在，orderId:%s", orderId));
    }
    if (order.getOrderStatus() != OrderStatusEnum.ORDER_PRE_PAY.getOrderStatus()) {
        throw new BusinessException(String.format("訂單狀態錯誤，order:%s", order));
    }

    // 2. 設定訂單為已取消狀態
    order.setOrderStatus((byte) OrderStatusEnum.ORDER_CLOSED_BY_EXPIRED.getOrderStatus());
    order.setUpdateTime(new Date());
    if (!orderService.updateById(order)) {
        throw new BusinessException("更新資料已失效");
    }

    // 3.商品貨品數量增加
    LambdaQueryWrapper<OrderItem> queryWrapper = Wrappers.lambdaQuery();
    queryWrapper.eq(OrderItem::getOrderId, orderId);
    List<OrderItem> orderItems = orderItemService.list(queryWrapper);
    for (OrderItem orderItem : orderItems) {
        if (orderItem.getSeckillId() != null) { // 秒殺單商品項處理
            Long seckillId = orderItem.getSeckillId();
            SeckillService seckillService = SpringContextUtil.getBean(SeckillService.class);
            if (!seckillService.addStock(seckillId)) {
                throw new BusinessException("秒殺商品貨品庫存增加失敗");

            }
        } else { // 普通單商品項處理
            Long goodsId = orderItem.getGoodsId();
            Integer goodsCount = orderItem.getGoodsCount();
            if (!goodsDao.addStock(goodsId, goodsCount)) {
                throw new BusinessException("秒殺商品貨品庫存增加失敗");
            }
        }
    }

    // 4. 返還優惠券
    couponService.releaseCoupon(orderId);
    log.info("---------------訂單orderId:{},未支付超時取消成功", orderId);
}

采用編程式事務對其優化，代碼如下：

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

public void doUnPaidTask(Long orderId) {
    // 啟用編程式事務
    // 1. 在開啟事務錢查詢訂單是否存在
    Order order = orderService.getById(orderId);
    if (order == null) {
        throw new BusinessException(String.format("訂單不存在，orderId:%s", orderId));
    }
    if (order.getOrderStatus() != OrderStatusEnum.ORDER_PRE_PAY.getOrderStatus()) {
        throw new BusinessException(String.format("訂單狀態錯誤，order:%s", order));
    }
    // 2. 開啟事務
    TransactionStatus transaction = platformTransactionManager.getTransaction(transactionDefinition);
    try {
        // 3. 設定訂單為已取消狀態
        order.setOrderStatus((byte) OrderStatusEnum.ORDER_CLOSED_BY_EXPIRED.getOrderStatus());
        order.setUpdateTime(new Date());
        if (!orderService.updateById(order)) {
            throw new BusinessException("更新資料已失效");
        }
        // 4. 商品貨品數量增加
        LambdaQueryWrapper<OrderItem> queryWrapper = Wrappers.lambdaQuery();
        queryWrapper.eq(OrderItem::getOrderId, orderId);
        List<OrderItem> orderItems = orderItemService.list(queryWrapper);
        for (OrderItem orderItem : orderItems) {
            if (orderItem.getSeckillId() != null) { // 秒殺單商品項處理
                Long seckillId = orderItem.getSeckillId();
                SeckillService seckillService = SpringContextUtil.getBean(SeckillService.class);
                RedisCache redisCache = SpringContextUtil.getBean(RedisCache.class);
                if (!seckillService.addStock(seckillId)) {
                    throw new BusinessException("秒殺商品貨品庫存增加失敗");
                }
                redisCache.increment(Constants.SECKILL_GOODS_STOCK_KEY + seckillId);
                redisCache.deleteCacheSet(Constants.SECKILL_SUCCESS_USER_ID + seckillId, order.getUserId());
            } else { // 普通單商品項處理
                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占用過高、記憶體溢位等，

這里給出海量日志高并發下優化點：

上報日志進行異步化處理，

普通版：采用阻塞佇列 ArrayBlockingQueue 得生產者消費者模式，對日志資料進行異步批量處理，在此場景下，通過生產者將資料快取再記憶體中，然后再消費者中批量保存入庫，
進階版：采用 Disruptor 佇列，也是基于記憶體佇列的生產者消費者模型，消費速度對比 ArrayBlockingQueue 有一個數量級得性能提升，附簡介說明：https://www.jianshu.com/p/bad7b4b44e48
終極版：采用 kfaka 訊息佇列中間件，持久日志資料，慢慢消費，雖然引入第三方依賴會增加系統復雜度，但是相比 kfaka 在大資料場景下提供的優秀表現，這一點也是值得，

如上三種方案：大家可以結合自身專案實際體量選擇

采集日志壓縮

對上報后的日志如果要再發送給其他服務，推薦是對其進行壓縮處理，避免消耗過多網路帶寬以及最終資料落庫選型：

網路傳輸，在 Java 里通常是指序列化方式，Jdk 自帶得序列化方式對比 Protobuf、fst、Hession 等在序列化速度和大小的表現上都沒有優勢，甚至可以用垃圾形容，博主這里直接給出 Java 得幾種序列化方式對比鏈接：https://segmentfault.com/a/1190000039934578，
建議對傳輸大小要求較高可以使用 Avro 序列化，對綜合要求較高可采用 Protobuf
落庫選型，像日志這種大資料量落庫，都是新增且無修改得場景建議使用 Clickhouse 進行存盤，相同資料量下對比 MySql 占用存盤更少，查詢性能更高

最后，附博主 github 地址：https://github.com/wayn111

歡迎大家點贊、收藏、轉發，你的支持將是博主更文的動力

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/539481.html

標籤：其他

上一篇：實體決議丨一文搞定GaussDB CM服務例外

下一篇：大資料-業務資料采集-FlinkCDC

大資料量、高并發業務怎么優化？（一）

一. 大資料量上傳寫入優化

1.1 如上功能描述很簡單，但是對于技術側想要做好這個功能，保證大用戶量（比如達到百萬級別）下，系統正常運行，功能正常其實是需要仔細思考的，博主這里給出思路：

1.2 批量寫入代碼優化

二. 大事務優化，減小影響范圍，提升系統處理能力

三. 客戶端海量日志上報優化