大家听起来这个项目是不是特别容易完成，用户量又少代表不需要考虑并发，也不会存在太大的流量冲击，性能要求也不会太高，似乎就是一个简简单单的单体增删改查项目呀？但事情远没有表面这么简单，诸位请接着往下看。

大家听起来这个项目是不是特别容易完成，用户量又少代表不需要考虑并发，也不会存在太大的流量冲击，性能要求也不会太高，似乎就是一个简简单单的单体增删改查项目呀？但事情远没有表面这么简单，诸位请接着往下看。

起初当我收到通知要负责这个需求时，从表面浅显的想了一下，似乎发现也不是太难，就是一个单体项目的 CRUD 工作，以我这手出神入化的 CV 大法， Hlod 住它简直轻轻松松，因此当时也没想太多就直接接手了，项目初期由于团队每位成员经验都很丰富，各自凭借着个人的 Copy 神功，项目的开发进度可谓是一骑千里，但慢慢的问题来了，而且这个问题还不小！

当时大概对外预计分发 1000 台机器，每台机器正式投入运营后，预估单日会产生 500~600 条数据的产出，套到前面的举例中，也就是大概会向几百个超市投放共计 1000 台收银机，每个收银台平均下来之后，大概单日内会有 500~600 个顾客结账！

当时大概对外预计分发 1000 台机器，每台机器正式投入运营后，预估单日会产生 500~600 条数据的产出，套到前面的举例中，也就是大概会向几百个超市投放共计 1000 台收银机，每个收银台平均下来之后，大概单日内会有 500~600 个顾客结账！

单月数据增长 1500W 的概念不言而喻，这也就代表着一年的数据增长量为 1500W * 12 = 1.8E ，这批机器投入后预估最少会运行三年起步，甚至十年乃至更久，同时第一批次就要投入 1000 台，后面可能还会有第二批次、第三批次.....的投入。

单月数据增长 1500W 的概念不言而喻，这也就代表着一年的数据增长量为 1500W * 12 = 1.8E ，这批机器投入后预估最少会运行三年起步，甚至十年乃至更久，同时第一批次就要投入 1000 台，后面可能还会有第二批次、第三批次.....的投入。

50W 只是最低的账单流水数据量，后续正式运营后可能数据量更大，此时架构的设计就成了难题！

1.2、方案的探讨

基本上当时团队的成员中，没人在此之前碰过这类需求，因此开了一个研讨会，去决定该如何将具体的方案落地，这里有人也许会说，数据量这么大，快上分布式/微服务啊！但实则解决不了这个问题， Why ？因为项目整体的用户量并不大，最多同一时刻也才 1000 并发请求，就算这个并发量再增大几个级别，这里用单体架构优化好了也能够抗住，所以问题并不在业务系统的架构上面，而是在数据落库这方面。

这里直接用分库可以吗？答案是也不行， Why ？因为整个项目中只有账单表才有这么大的数据量，其他的用户表、系统表、功能菜单表、后台表......，基本上不会有太大的数据量，所以直接做分库也没必要，属实有些浪费资源。

这里直接用分库可以吗？答案是也不行， Why ？因为整个项目中只有账单表才有这么大的数据量，其他的用户表、系统表、功能菜单表、后台表......，基本上不会有太大的数据量，所以直接做分库也没必要，属实有些浪费资源。

有小伙伴可能想到了！可以按月份对流水表做分区呀！乍一听似乎像那么一回事，但依旧不行，因为第一批机器投入后，单月预计就会产生 1500W 条数据，后续可能会增加机器数量，因此单月的数据量达到 2000W、3000W..... 都有可能，如果按月做表分区，每个分区里面都有几千万条数据，一张账单表的流水随着时间推移，数据量甚至会达到几十亿！

一张表中存储几十亿条数据，这基本上不现实，虽然 InnoDB 在数据页为 16KB 尺寸下，单表最多能存储 64TB 数据，有可能这几十亿条数据真的能存下去，但查询时的性能简直令人头大，并且最关键的是不方便后续对数据做维护、管理、备份和迁移工作。

一张表中存储几十亿条数据，这基本上不现实，虽然 InnoDB 在数据页为 16KB 尺寸下，单表最多能存储 64TB 数据，有可能这几十亿条数据真的能存下去，但查询时的性能简直令人头大，并且最关键的是不方便后续对数据做维护、管理、备份和迁移工作。

最终架构定型为：业务系统使用单体架构 + 数据库使用单库 + 流水表按月份做水平分表。

最终架构定型为：业务系统使用单体架构 + 数据库使用单库 + 流水表按月份做水平分表。

在上一阶段中已经决定好了具体的方案，但又该如何将方案落地呢？首先咱们先把方案落地的思路捋清楚：

①能够自动按月创建一张月份账单表，从而将每月的流水数据写入进去。

②写入数据时，能够根据当前的日期，选择对应的月份账单表并插入数据。

①能够自动按月创建一张月份账单表，从而将每月的流水数据写入进去。

②写入数据时，能够根据当前的日期，选择对应的月份账单表并插入数据。

实现了上面两个需求后，整个方案近乎落地了一半，但接下来该如何去实现相应功能呢？咱们一点点来动手实现。

2.1、利用存储过程实现按月动态创建表

创建表的 SQL 语句大家都不陌生，按月份创建表之前，自然也需要一份原生创建表的 DDL 语句，如下：

CREATE TABLE `month_bills_202211` (

`month_bills_id` int(8) NOT NULL AUTO_INCREMENT COMMENT '账单ID',

`serial_number` varchar(50) NOT NULL COMMENT '流水号',

`bills_info` text NOT NULL COMMENT '账单详情',

`pay_money` decimal(10,3) NOT NULL COMMENT '支付金额',

`machine_serial_no` varchar(20) NOT NULL COMMENT '收银机器',

`bill_date` timestamp NOT NULL COMMENT '账单日期',

`bill_comment` varchar(100) NULL DEFAULT '无'COMMENT '账单备注',

PRIMARY KEY (`month_bills_id`) USING BTREE,

UNIQUE `serial_number` (`serial_number`),

KEY `bill_date` (`bill_date`)

)

ENGINE = InnoDB

CHARACTER SET = utf8

COLLATE = utf8_general_ci

ROW_FORMAT = Compact;

上述的语句会创建一张月份账单表，这张表主要包含七个字段，如下：

其中注意的几个小细节：

①日期字段使用的是 timestamp 类型，而并非 datetime ，因为前者更省空间。

②账单详情字段用的是 text 类型，因为这个字段可能会出现很多的信息。

③定义了一个和表没有关系的自增字段作为主键，用于维护聚簇索引树的结构。

①日期字段使用的是 timestamp 类型，而并非 datetime ，因为前者更省空间。

②账单详情字段用的是 text 类型，因为这个字段可能会出现很多的信息。

③定义了一个和表没有关系的自增字段作为主键，用于维护聚簇索引树的结构。

除开有上述七个字段外，还有三个索引：

到这里就有了最基本的建表语句，主要是用来创建第一张月份账单表，如果想要实现动态按照每月建表的话，还需要用到存储过程来实现，接着来写一个存储过程。

最终撰写出的存储过程如下：

DELIMITER //

DROP PROCEDURE IF EXISTS create_table_by_month //

CREATE PROCEDURE `create_table_by_month`

BEGIN

-- 用于记录下一个月份是多久

DECLARE nextMonth varchar(20);

-- 用于记录创建表的SQL语句

DECLARE createTableSQL varchar(5210);

-- 执行创建表的SQL语句后，获取表的数量

DECLARE tableCount int;

-- 用于记录要生成的表名

DECLARE tableName varchar(20);

-- 用于记录表的前缀

DECLARE table_prefix varchar(20);

-- 获取下个月的日期并赋值给nextMonth变量

SELECT SUBSTR(

replace(

DATE_ADD(CURDATE, INTERVAL 1 MONTH),

'-', ''),

1, 6) INTO @nextMonth;

-- 设置表前缀变量值为td_user_banks_log_

set@table_prefix = 'month_bills_';

-- 定义表的名称=表前缀+月份，即 month_bills_2022112 这个格式

SET @tableName = CONCAT(@table_prefix, @nextMonth);

-- 定义创建表的SQL语句

set@createTableSQL=concat( "create table if not exists ",@tableName, "(

`month_bills_id` int(8) NOT NULL AUTO_INCREMENT COMMENT '账单ID',

`serial_number` varchar(50) NOT NULL COMMENT '流水号',

`bills_info` text NOT NULL COMMENT '账单详情',

`pay_money` decimal(10,3) NOT NULL COMMENT '支付金额',

`machine_serial_no` varchar(20) NOT NULL COMMENT '收银机器',

`bill_date` timestamp NOT NULL DEFAULT now COMMENT '账单日期',

`bill_comment` varchar(100) NULL DEFAULT '无' COMMENT '账单备注',

PRIMARY KEY (`month_bills_id`) USING BTREE,

UNIQUE `serial_number` (`serial_number`),

KEY `bill_date` (`bill_date`)

) ENGINE = InnoDB

CHARACTER SET = utf8

COLLATE = utf8_general_ci

ROW_FORMAT = Compact;" );

-- 使用 PREPARE 关键字来创建一个预备执行的SQL体

PREPARE create_stmt from @createTableSQL;

-- 使用 EXECUTE 关键字来执行上面的预备SQL体：create_stmt

EXECUTE create_stmt;

-- 释放掉前面创建的SQL体（减少内存占用）

DEALLOCATE PREPARE create_stmt;

-- 执行完建表语句后，查询表数量并保存再 tableCount 变量中

SELECT

COUNT(1) INTO @tableCount

FROM

information_schema.`TABLES`

WHERE TABLE_NAME = @tableName;

-- 查询一下对应的表是否已存在

SELECT @tableCount 'tableCount';

END //

delimiter ;

上述这个存储过程比较长，但基本上都写好了注释，所以阅读起来应该还是比较轻松的，也包括该存储过程在 MySQL5.1、8.0 版本中都测试过，所以大家也可以直接用，主要拆解一下里面较为难理解的一句 SQL ，如下：

SELECT SUBSTR(

replace(

DATE_ADD(CURDATE, INTERVAL 1 MONTH),

'-', ''),

1, 6) INTO @nextMonth;

-- 在当前日期的基础上增加一个月，如2022-11-12 23:46:11，会得到2022-12-12 23:46:11

select DATE_ADD(CURDATE, INTERVAL 1 MONTH);

-- 使用空字符代替日期中的 - 符号，得到 20221212 23:46:11 这样的效果

select replace( '2022-12-12 23:46:11', '-', '');

-- 对字符串做截取，获取第一位到第六位，得到 202212 这样的效果

select SUBSTR( "20221212 23:46:11",1,6);

经过上述拆解之后大家应该能看的很清楚，最终每次调用该存储过程时，都会基于当前数据库的时间，然后向后增加一个月，同时将格式转化为 YYYYMM 格式，接下来调用该存储过程，如下：

call create_table_by_month;

+------------+

| tableCount |

+------------+

| 1 |

+------------+

当返回的值为 1 而并非 0 时，就表示已经在数据库中查到了前面通过存储过程创建的表，即表示动态创建表的存储过程可以生效！接着为了能够每月定时触发，可以在 MySQL 中注册一个每月执行一次的定时事件，如下：

create EVENT

`create_table_by_month_event` -- 创建一个定时器

ON SCHEDULE EVERY

1 MONTH -- 每间隔一个月执行一次

STARTS

'2022-11-28 00:00:00'-- 从2022-11-28 00:00:00后开始

ON COMPLETION

PRESERVE ENABLE -- 执行完成之后不删除定时器

DO

call create_table_by_month; -- 每次触发定时器时执行的语句

MySQL5.1 版本中除开引入了存储过程/函数、触发器的支持外，还引入了定时器的技术，也就是支持定时执行一条 SQL ，此时咱们可借助 MySQL 自带的定时器来定时调用之前的存储过程，最终实现按月定时创建表的需求！

但定时器在使用之前，需要先查看定时器是否开启，如下： show variables like 'event_scheduler'; 如果是 OFF 关闭状态，需要通过 set global event_scheduler = 1 | on; 命令开启。如果想要永久生效， MySQL8.0 以下的版本可找到 my.ini/my.conf 文件，然后找到 [mysqld] 的区域，再里面多加入一行 event_scheduler = ON 的配置即可。

这里再附上一些管理定时器的命令：

-- 查看创建的定时器

show events;

select * from mysql.event;

select * from information_schema.EVENTS;

-- 删除一个定时器

drop event 定时器名称;

-- 关闭一个定时器任务

alter event 定时器名称 on COMPLETION PRESERVE DISABLE;

-- 开启一个定时器任务

alter event 定时器名称 on COMPLETION PRESERVE ENABLE;

2.2、写入数据时能够根据月份插入对应表

作为一个后端项目，必然还需要搭建客户端，这里用 SpringBoot+MyBatis 来快速构建一个单体项目（最后会给出完整源码），这里需要注意，月份账单表对应的实体类中要多出一个 targetTable 字段，如下：

public class MonthBills {

// 月份账单表ID

private Integer monthBillsId;

// 账单流水号

private String serialNumber;

// 支付金额

private BigDecimal payMoney;

// 收银机器

private String machineSerialNo;

// 账单日期

private Date billDate;

// 账单详情

private String billsInfo;

// 账单备注

private String billComment;

// 要操作的目标表

private String targetTable;

// 省略构造方法和Get/Set方法.....

}

上述的实体类与之前的表字段结构几乎完全相同，但会多出一个 targetTable 属性，后续会用来记录要操作的目标表，接着再撰写一个工具类，如下：

public class TableTimeUtils {

/*

* 使用ThreadLocal来确保线程安全，或者可以使用Java8新引入的DateTimeFormatter类：

* monthTL：负责将一个日期处理成 YYYYMM 格式

*/

private static ThreadLocal monthTL =

ThreadLocal.withInitial( ->

new SimpleDateFormat( "YYYYMM"));

// 表的前缀

private static String tablePrefix = "month_bills_";

// 将一个日期格式化为YYYYMM格式

public static String getYearMonth(Date date) {

returnmonthTL.get.format(date);

}

// 获取目标数据的表名（操作单条数据公用的方法：增删改查）

public static void getDataByTable(MonthBills monthBills){

// 获取传入对象的时间

Date billDate = monthBills.getBillDate;

// 根据该对象中的时间，计算出要操作的表名后缀

String yearMonth = getYearMonth(billDate);

// 将表前缀和后缀拼接，得到完整的表名，如：month_bills_202211

monthBills.setTargetTable(tablePrefix + yearMonth);

}

}

这个工具类主要负责处理日期的时间格式，以及用来定位要操作的目标表名，对于日期格式化类： SimpleDateFormat 由于是线程不安全的，所以使用 ThreadLocal 来确保线程安全！上述工具类中主要提供了两个基础方法：

getYearMonth ：将一个日期格式化成 YYYYMM 格式。

getDataByTable ：获取单条数据操作时的表名。

getYearMonth ：将一个日期格式化成 YYYYMM 格式。

getDataByTable ：获取单条数据操作时的表名。

有了工具类后，接着来撰写 Dao、Mapper 层的代码，如下：

@Mapper

@Repository

public interface MonthBillsMapper {

int deleteByPrimaryKey(Integer monthBillsId);

int insertSelective(MonthBills record);

MonthBills selectByPrimaryKey(Integer monthBillsId);

int updateByPrimaryKeySelective(MonthBills record);

}

上述是月份账单表对应的 Dao/Mapper 接口，因为我这里是通过 MyBatis 的逆向工程文件自动生成的，所以名字就是上面那样，我这边未成更改，接着来看看对应的 xml 文件，如下：

insert into ${targetTable}

< iftest= "monthBillsId != null">

month_bills_id,

< iftest= "serialNumber != null">

serial_number,

< iftest= "payMoney != null">

pay_money,

< iftest= "machineSerialNo != null">

machine_serial_no,

< iftest= "billDate != null">

bill_date,

< iftest= "billComment != null">

bill_comment,

< iftest= "billsInfo != null">

bills_info,

< iftest= "monthBillsId != null">

#{monthBillsId,jdbcType=INTEGER},

< iftest= "serialNumber != null">

#{serialNumber,jdbcType=VARCHAR},

< iftest= "payMoney != null">

#{payMoney,jdbcType=DECIMAL},

< iftest= "machineSerialNo != null">

#{machineSerialNo,jdbcType=VARCHAR},

< iftest= "billDate != null">

#{billDate,jdbcType=TIMESTAMP},

< iftest= "billComment != null">

#{billComment,jdbcType=VARCHAR},

< iftest= "billsInfo != null">

#{billsInfo,jdbcType=LONGVARCHAR},

上述这么大一长串，其实也不是俺手敲的，依旧是 MyBatis 逆向工程生成的代码，但我对其中的一处稍微做了改动，如下：

-- 原本生成的代码是：

insert into month_bills_202211

-- 然后被我改成了：

insert into ${targetTable}

还记得最开始的实体类中，咱们多添加的那个 targetTable 属性嘛？在这里会根据该字段的值动态的去操作不同月份的表，接着来写一下 Service 层的接口和实现类，如下：

// Service接口（目前里面只有一个方法）

public interface IMonthBillsService {

int insert(MonthBills monthBills);

}

// Service实现类

@Service

public class MonthBillsServiceImpl implements IMonthBillsService {

@Autowired

private MonthBillsMapper billsMapper;

@Override

public int insert(MonthBills monthBills) {

// 获取要插入数据的表名

TableTimeUtils.getDataByTable(monthBills);

// 返回插入数据的状态

returnbillsMapper.insertSelective(monthBills);

}

}

在 service 层目前仅实现了一个插入数据的方法，其中的逻辑也非常简单，仅仅在调用 Dao 层的插入方法之前，获取了一下当前这条数据要插入的表名，最后来看看 Controller/API 层，如下：

@RestController

@RequestMapping( "/bills")

public class MonthBillsAPI {

@Autowired

private IMonthBillsService billsService;

// 账单结算的API

@RequestMapping( "/settleUp")

public String settleUp(MonthBills monthBills){

// 设置账单交易时间为当前时间

monthBills.setBillDate(new Date(System.currentTimeMillis));

// 使用UUID随机生成一个流水号

monthBills.setSerialNumber(monthBills.getMachineSerialNo

+ System.currentTimeMillis);

// 调用新增账单数据的service方法

if(billsService.insert(monthBills) > 0){

return">>>>账单结算成功>>>账单结算失败