qidao123.com ToB IT社区-企服评测·应用市场

标题: 大数据量、高并发业务怎么优化？（一） [打印本页]

---

作者: 立聪堂德州十三局店    时间: 2022-12-15 07:22
标题: 大数据量、高并发业务怎么优化？（一）
博主这里的大数据量、高并发业务处理优化基于博主线上项目实践以及全网资料整理而来，在这里分享给大家
一. 大数据量上传写入优化

线上业务后台项目有一个消息推送的功能，通过上传包含用户id的文件，给指定用户推送系统消息

1.1 如上功能描述很简单，但是对于技术侧想要做好这个功能，保证大用户量（比如达到百万级别）下，系统正常运行，功能正常其实是需要仔细思考的，博主这里给出思路：

• 上传文件类型选择
  

通常情况下大部分用户都会使用excel文件，但是相比excel文件还有一种更加推荐的文件格式，那就是csv文件，相比excel文件它可以直接在记事本编辑，excel也可以打开cvs文件，且占用内存更少（画重点），对于上传的csv文件过于庞大，也可以采用流式读取，读一部分写一部分

• 消息推送成功与否状态保存
  

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

• 批量写入启不启用事务
  

博主这里给出两种方案利弊：

• 启用事务：好处在于如批量插入过程中，异常情况可以保证原子性，但是性能比不开事务低，在特大数据量下会明显低一个档次
  
• 不启用事务：好处就是写入性能高，特大数据量写入性能提升明显，但是无法保证原子性，但是对于已经批量插入的新增数据，只是会产生脏数据而已，在功能设计合理的情况下是不影响业务的，如下面第四点
  

综上：在大数据量下，我们要是追求极致性能可以不启用事务，具体选择也需各位结合自身业务情况

• 推送异常失败的消息处理
  

建议功能设计上，可以屏蔽对失败消息再进行操作，这样不需要再处理之前推送失败写入的脏数据，直接新增消息推送即可
1.2 批量写入代码优化

• jdbc参数携带 rewriteBatchedStatements=true 在jdbc驱动上启动批量写入功能，如下
  

1. 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问题不大，字段多了需要降低这个写入量
  

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

复制代码

一般情况下大家都知道第二条优化，但是可能会忽略jdbc参数携带 rewriteBatchedStatements=true，这个参数能在第二条的基础上启用批量执行SQL，进一步提升写入性能
二. 大事务优化，减小影响范围，提升系统处理能力

@Transactional 大于 Spring 提供得事务注解，许多人都知道，但是在高并发下，不建议使用，推荐通过编程式事务来手动控制事务提交或者回滚，减少事务影响范围
如下是一段订单超时未支付回滚业务数据得代码，采用 @Transactional 事务注解

1. @Transactional(rollbackFor = Exception.class)
2. public void doUnPaidTask(Long orderId) {
3.     // 1. 查询订单是否存在
4.     Order order = orderService.getById(orderId);
5.     if (order == null) {
6.         throw new BusinessException(String.format("订单不存在，orderId:%s", orderId));
7.     }
8.     if (order.getOrderStatus() != OrderStatusEnum.ORDER_PRE_PAY.getOrderStatus()) {
9.         throw new BusinessException(String.format("订单状态错误，order:%s", order));
10.     }
12.     // 2. 设置订单为已取消状态
13.     order.setOrderStatus((byte) OrderStatusEnum.ORDER_CLOSED_BY_EXPIRED.getOrderStatus());
14.     order.setUpdateTime(new Date());
15.     if (!orderService.updateById(order)) {
16.         throw new BusinessException("更新数据已失效");
17.     }
19.     // 3.商品货品数量增加
20.     LambdaQueryWrapper<OrderItem> queryWrapper = Wrappers.lambdaQuery();
21.     queryWrapper.eq(OrderItem::getOrderId, orderId);
22.     List<OrderItem> orderItems = orderItemService.list(queryWrapper);
23.     for (OrderItem orderItem : orderItems) {
24.         if (orderItem.getSeckillId() != null) { // 秒杀单商品项处理
25.             Long seckillId = orderItem.getSeckillId();
26.             SeckillService seckillService = SpringContextUtil.getBean(SeckillService.class);
27.             if (!seckillService.addStock(seckillId)) {
28.                 throw new BusinessException("秒杀商品货品库存增加失败");
30.             }
31.         } else { // 普通单商品项处理
32.             Long goodsId = orderItem.getGoodsId();
33.             Integer goodsCount = orderItem.getGoodsCount();
34.             if (!goodsDao.addStock(goodsId, goodsCount)) {
35.                 throw new BusinessException("秒杀商品货品库存增加失败");
36.             }
37.         }
38.     }
40.     // 4. 返还优惠券
41.     couponService.releaseCoupon(orderId);
42.     log.info("---------------订单orderId:{},未支付超时取消成功", orderId);
43. }

复制代码

采用编程式事务对其优化，代码如下：

1. @Resource
2. private PlatformTransactionManager platformTransactionManager;
3. @Resource
4. private TransactionDefinition transactionDefinition;
6. public void doUnPaidTask(Long orderId) {
7.     // 启用编程式事务
8.     // 1. 在开启事务钱查询订单是否存在
9.     Order order = orderService.getById(orderId);
10.     if (order == null) {
11.         throw new BusinessException(String.format("订单不存在，orderId:%s", orderId));
12.     }
13.     if (order.getOrderStatus() != OrderStatusEnum.ORDER_PRE_PAY.getOrderStatus()) {
14.         throw new BusinessException(String.format("订单状态错误，order:%s", order));
15.     }
16.     // 2. 开启事务
17.     TransactionStatus transaction = platformTransactionManager.getTransaction(transactionDefinition);
18.     try {
19.         // 3. 设置订单为已取消状态
20.         order.setOrderStatus((byte) OrderStatusEnum.ORDER_CLOSED_BY_EXPIRED.getOrderStatus());
21.         order.setUpdateTime(new Date());
22.         if (!orderService.updateById(order)) {
23.             throw new BusinessException("更新数据已失效");
24.         }
25.         // 4. 商品货品数量增加
26.         LambdaQueryWrapper<OrderItem> queryWrapper = Wrappers.lambdaQuery();
27.         queryWrapper.eq(OrderItem::getOrderId, orderId);
28.         List<OrderItem> orderItems = orderItemService.list(queryWrapper);
29.         for (OrderItem orderItem : orderItems) {
30.             if (orderItem.getSeckillId() != null) { // 秒杀单商品项处理
31.                 Long seckillId = orderItem.getSeckillId();
32.                 SeckillService seckillService = SpringContextUtil.getBean(SeckillService.class);
33.                 RedisCache redisCache = SpringContextUtil.getBean(RedisCache.class);
34.                 if (!seckillService.addStock(seckillId)) {
35.                     throw new BusinessException("秒杀商品货品库存增加失败");
36.                 }
37.                 redisCache.increment(Constants.SECKILL_GOODS_STOCK_KEY + seckillId);
38.                 redisCache.deleteCacheSet(Constants.SECKILL_SUCCESS_USER_ID + seckillId, order.getUserId());
39.             } else { // 普通单商品项处理
40.                 Long goodsId = orderItem.getGoodsId();
41.                 Integer goodsCount = orderItem.getGoodsCount();
42.                 if (!goodsDao.addStock(goodsId, goodsCount)) {
43.                     throw new BusinessException("秒杀商品货品库存增加失败");
44.                 }
45.             }
46.         }
48.         // 5. 返还优惠券
49.         couponService.releaseCoupon(orderId);
50.         // 6. 所有更新操作完成后，提交事务
51.         platformTransactionManager.commit(transaction);
52.         log.info("---------------订单orderId:{},未支付超时取消成功", orderId);
53.     } catch (Exception e) {
54.         log.info("---------------订单orderId:{},未支付超时取消失败", orderId, e);
55.         // 7. 发生异常，回滚事务
56.         platformTransactionManager.rollback(transaction);
57.     }
58. }

复制代码

可以看到采用编程式事务后，我们将查询逻辑排除在事务之外，减小了其影响范围，也就提升了性能，在高并发场景下，性能优先的场景，我们甚至可以考虑不适用事务
三. 客户端海量日志上报优化

线上项目客户端，采用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

欢迎大家点赞、收藏、转发，你的支持将是博主更文的动力

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

---

欢迎光临 qidao123.com ToB IT社区-企服评测·应用市场 (https://dis.qidao123.com/)

Powered by Discuz! X3.5