Spring Boot實戰解決高併發數據入庫: Redis 快取+MySQL 批次入庫

前言

最近在做閱讀類的業務，需要記錄使用者的PV，UV；

專案狀況：前期嘗試業務階段；

特點：

• 快速實現（不需要做太重，滿足初期推廣運營即可）
• 快速投入市場去運營

收集使用者的原始數據，三要素：

• 誰
• 在什麼時間
• 閱讀哪篇文章

提到PV，UV腦海中首先浮現特點：

• 需要考慮效能（每個客戶每開啓一篇文章進行記錄）
• 允許數據有較小誤差（少部分數據丟失）

架構設計

• 架構圖：

• 時序圖
  
• 記錄基礎數據MySQL表結構

CREATE TABLE `zh_article_count` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `bu_no` varchar(32) DEFAULT NULL COMMENT '業務編碼',
  `customer_id` varchar(32) DEFAULT NULL COMMENT '使用者編碼',
  `type` int(2) DEFAULT '0' COMMENT '統計型別:0APP內文章閱讀',
  `article_no` varchar(32) DEFAULT NULL COMMENT '文章編碼',
  `read_time` datetime DEFAULT NULL COMMENT '閱讀時間',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '建立時間',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '更新時間',
  `param1` int(2) DEFAULT NULL COMMENT '預留欄位1',
  `param2` int(4) DEFAULT NULL COMMENT '預留欄位2',
  `param3` int(11) DEFAULT NULL COMMENT '預留欄位3',
  `param4` varchar(20) DEFAULT NULL COMMENT '預留欄位4',
  `param5` varchar(32) DEFAULT NULL COMMENT '預留欄位5',
  `param6` varchar(64) DEFAULT NULL COMMENT '預留欄位6',
  PRIMARY KEY (`id`) USING BTREE,
  UNIQUE KEY `uk_zh_article_count_buno` (`bu_no`),
  KEY `key_zh_article_count_csign` (`customer_id`),
  KEY `key_zh_article_count_ano` (`article_no`),
  KEY `key_zh_article_count_rtime` (`read_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='文章閱讀統計表';

• 技術實現方案

  SpringBoot

  Redis

  MySQL

程式碼實現

• 完整程式碼（GitHub，歡迎大家Star,Fork,Watch)

  https://github.com/dangnianchuntian/redis-to-db

• 主要程式碼展示

  • Controller

  /*
   * Copyright (c) 2020. [email protected] All Rights Reserved.
   * 專案名稱：Spring Boot實戰解決高併發數據入庫: Redis 快取+MySQL 批次入庫
   * 類名稱：RedisConfig.java
   * 建立人：張晗
   * 聯繫方式：[email protected]
   * 開源地址: https://github.com/dangnianchuntian/redis-to-db
   * 部落格地址: https://zhanghan.blog.csdn.net
   */
  
  package com.zhanghan.redistodb.controller;
  
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.validation.annotation.Validated;
  import org.springframework.web.bind.annotation.RequestBody;
  import org.springframework.web.bind.annotation.RequestMapping;
  import org.springframework.web.bind.annotation.RequestMethod;
  import org.springframework.web.bind.annotation.RestController;
  
  import com.zhanghan.redistodb.controller.request.PostArticleViewsRequest;
  import com.zhanghan.redistodb.service.ArticleCountService;
  
  @RestController
  public class ArticleCountController {
  
      @Autowired
      private ArticleCountService articleCountService;
  
     /**
      * 記錄使用者存取記錄
      */
      @RequestMapping(value = "/post/article/views", method = RequestMethod.POST)
      public Object postArticleViews(@RequestBody @Validated PostArticleViewsRequest postArticleViewsRequest) {
          return articleCountService.postArticleViews(postArticleViewsRequest);
      }
  
      /**
       *  批次將快取中的數據同步到MySQL（模擬定時任務操作）
       */
      @RequestMapping(value = "/post/batch", method = RequestMethod.POST)
      public Object postBatch() {
          return articleCountService.postBatchRedisToDb();
      }
  
  }
  

  • Service

/*
 * Copyright (c) 2020. [email protected] All Rights Reserved.
 * 專案名稱：Spring Boot實戰解決高併發數據入庫: Redis 快取+MySQL 批次入庫
 * 類名稱：RedisConfig.java
 * 建立人：張晗
 * 聯繫方式：[email protected]
 * 開源地址: https://github.com/dangnianchuntian/redis-to-db
 * 部落格地址: https://zhanghan.blog.csdn.net
 */

package com.zhanghan.redistodb.service.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import com.alibaba.fastjson.JSON;
import com.zhanghan.redistodb.controller.request.PostArticleViewsRequest;
import com.zhanghan.redistodb.dto.ArticleCountDto;
import com.zhanghan.redistodb.mybatis.mapper.XArticleCountMapper;
import com.zhanghan.redistodb.service.ArticleCountService;
import com.zhanghan.redistodb.util.wrapper.WrapMapper;

import cn.hutool.core.util.IdUtil;

@Service
public class ArticleCountServiceImpl implements ArticleCountService {

    private static Logger logger = LoggerFactory.getLogger(ArticleCountServiceImpl.class);

    @Autowired
    private RedisTemplate<String, String> strRedisTemplate;

    @Autowired
    private XArticleCountMapper xArticleCountMapper;

    @Value("${zh.article.count.redis.key:zh}")
    private String zhArticleCountRedisKey;

    @Value("#{T(java.lang.Integer).parseInt('${zh..article.read.num:3}')}")
    private Integer articleReadNum;

    /**
     * 記錄使用者存取記錄
     */
    @Override
    public Object postArticleViews(PostArticleViewsRequest postArticleViewsRequest) {

        ArticleCountDto articleCountDto = new ArticleCountDto();
        articleCountDto.setBuNo(IdUtil.simpleUUID());
        articleCountDto.setCustomerId(postArticleViewsRequest.getCustomerId());
        articleCountDto.setArticleNo(postArticleViewsRequest.getArticleNo());
        articleCountDto.setReadTime(new Date());
        String strArticleCountDto = JSON.toJSONString(articleCountDto);
        strRedisTemplate.opsForList().rightPush(zhArticleCountRedisKey, strArticleCountDto);
        return WrapMapper.ok();
    }

    /**
     * 批次將快取中的數據同步到MySQL
     */
    @Override
    public Object postBatchRedisToDb() {

        Date now = new Date();

        while (true) {

            List<String> strArticleCountList =
                    strRedisTemplate.opsForList().range(zhArticleCountRedisKey, 0, articleReadNum);

            if (CollectionUtils.isEmpty(strArticleCountList)) {
                return WrapMapper.ok();
            }

            List<ArticleCountDto> articleCountDtoList = new ArrayList<>();

            strArticleCountList.stream().forEach(x -> {
                ArticleCountDto articleCountDto = JSON.parseObject(x, ArticleCountDto.class);
                articleCountDtoList.add(articleCountDto);
            });

            //過濾出本次定時任務之前的快取中數據,防止死回圈
            List<ArticleCountDto> beforeArticleCountDtoList = articleCountDtoList.stream().filter(x -> x.getReadTime()
                    .before(now)).collect(Collectors.toList());

            if (CollectionUtils.isEmpty(beforeArticleCountDtoList)) {
                return WrapMapper.ok();
            }

            xArticleCountMapper.batchAdd(beforeArticleCountDtoList);

            Integer delSize = beforeArticleCountDtoList.size();

            strRedisTemplate.opsForList().trim(zhArticleCountRedisKey, delSize, -1L);
        }

    }

}

測試

• 模擬使用者請求存取後臺（多次請求）
  
• 檢視快取中存取數據
  
• 模擬定時任務將快取中數據同步到DB中
  
• 這時檢視快取中的數據已經沒了
  
• 檢視數據庫表結構
  

總結

• 專案中定時任務
  • 問演示方便用http代替定時任務排程；實際專案中用XXL-job,參考:定時任務的選型及改造
  • 定時任務專案中用redis鎖防止併發（定時任務排程端多次排程等)，參考:Redis實現計數器—介面防刷—升級版（Redis+Lua）
• 後期運營數據可以從閱讀記錄表中拉數據進行相關分析
• 存取量大：可以將MySQL中的閱讀記錄表定時遷移走（MySQL建歷史表，MongoDB等）