博主這裡的大數據量、高併發業務處理優化基於博主線上項目實踐以及全網資料整理而來，在這裡分享給大家一. 大數據量上傳寫入優化線上業務後臺項目有一個消息推送的功能，通過上傳包含用戶id的文件，給指定用戶推送系統消息 1.1 如上功能描述很簡單，但是對於技術側想要做好這個功能，保證大用戶量（比如達到百 ...

博主這裡的大數據量、高併發業務處理優化基於博主線上項目實踐以及全網資料整理而來，在這裡分享給大家

一. 大數據量上傳寫入優化

線上業務後臺項目有一個消息推送的功能，通過上傳包含用戶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

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

大數據量、高併發業務怎麼優化？（一）

一. 大數據量上傳寫入優化

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

1.2 批量寫入代碼優化

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

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