MySQL 千万级数据处理_mysql批量更新千万数据

---

一、优化

单表优化

除非单表数据未来会一直不断上涨，否则不要一开始就考虑拆分，拆分会带来逻辑、部署、运维的各种复杂度，一般以整型值为主的表在千万级以下，字符串为主的表在五百万以下是没有太大问题的。而事实上很多时候 MySQL 单表的性能依然有不少优化空间，甚至能正常支撑千万级以上的数据量：

A、字段

1、尽量使用 TINYINT、SMALLINT、MEDIUM_INT 作为整数类型而非 INT，如果非负则加上UNSIGNED

2、VARCHAR 的长度只分配真正需要的空间

3、使用枚举或整数代替字符串类型

4、尽量使用 TIMESTAMP 而非 DATETIME，

5、单表不要有太多字段，建议在 20 以内

6、避免使用 NULL 字段，很难查询优化且占用额外索引空间

7、用整型来存 IP

B、索引

索引的种类：

1、主键索引 (把某列设为主键，则就是主键索引)

2、唯一索引(unique) (即该列具有唯一性，同时又是索引)

3、index （普通索引）

4、全文索引(FULLTEXT)

5、复合索引(多列和在一起)

索引建立注意事项：

1、索引并不是越多越好，要根据查询有针对性的创建，考虑在WHERE和ORDER BY命令上涉及的列建立索引，可根据 EXPLAIN 来查看是否用了索引还是全表扫描

2、应尽量避免在 WHERE 子句中对字段进行 NULL 值判断，否则将导致引擎放弃使用索引而进行全表扫描如：select id from t where num is null 可以在 num 上设置默认值 0，确保表中 num 列没有 null 值，然后这样查询：select id from t where num = 0

3、值分布很稀少的字段不适合建索引，例如 “性别” 这种只有两三个值的字段

4、字符字段只建前缀索引

5、并不是所有索引对查询都有效，SQL 是根据表中数据来进行查询优化的，当索引列有大量数据重复时，SQL查询可能不会去利用索引，如一表中有字段 sex，male、female 几乎各一半，那么即使在sex上建了索引也对查询效率起不了作用。

6、应尽可能的避免更新 clustered 索引数据列，因为 clustered 索引数据列的顺序就是表记录的物理存储顺序，一旦该列值改变将导致整个表记录的顺序的调整，会耗费相当大的资源。若应用系统需要频繁更新 clustered 索引数据列，那么需要考虑是否应将该索引建为 clustered 索引。

7、应尽量避免在 where 子句中使用 != 或 <> 操作符，否则将引擎放弃使用索引而进行全表扫描。

8、尽量避免在 where 子句中使用 or 来连接条件，否则将导致引擎放弃使用索引而进行全表扫描，如：select id from t where num = 10 or num = 20

可以这样查询：

select id from t where num = 10 

union all 

select id from t where num = 20
1
2
3
4
5

9、下面的查询也将导致全表扫描：select id from t where name like ‘%abc%’ 若要提高效率，可以考虑全文检索。

10、in 和 not in 也要慎用，否则会导致全表扫描，如：select id from t where num in(1,2,3)

对于连续的数值，能用 between 就不要用 in 了：

select id from t where num between 1 and 3
1

11、如果在 where 子句中使用参数，也会导致全表扫描。因为 SQL 只有在运行时才会解析局部变量，但优化程序不能将访问计划的选择推迟到运行时。它必须在编译时进行选择。然而，如果在编译时建立访问计划，变量的值还是未知的，因而无法作为索引选择的输入项。如下面语句将进行全表扫描：select id from t where num = @num

可以改为强制查询使用索引：

select id from t with(index(索引名)) where num = @num
1

12、不要在 where 子句中的 “=” 左边进行函数、算术运算或其他表达式运算，否则系统将可能无法正确使用索引。如：select id from t where num / 2 = 100

select id from t where substring(name, 1 ,3) = ’abc’ # 查询name以abc开头的id列表
1

分别应改为:

select id from t where num = 100 * 2

select id from t where name like ‘abc%’ 
1
2
3

13、在使用索引字段作为条件时，如果该索引是复合索引，那么必须使用到该索引中的第一个字段作为条件时才能保证系统使用该索引，否则该索引将不会被使用，并且应尽可能的让字段顺序与索引顺序相一致。

14、很多时候用 exists 代替 in 是一个好的选择：select num from a where num in(select num from b)

用下面的语句替换：

select num from a where exists(select 1 from b where num = a.num)
1

C、引擎

目前广泛使用的是 MyISAM 和 InnoDB 两种引擎：

MyISAM：

MyISAM引擎是MySQL 5.1及之前版本的默认引擎，它的特点是：

• 不支持行锁，读取时对需要读到的所有表加锁，写入时则对表加排它锁
• 不支持事务
• 不支持外键
• 不支持崩溃后的安全恢复
• 在表有读取查询的同时，支持往表中插入新纪录
• 支持BLOB和TEXT的前500个字符索引，支持全文索引
• 支持延迟更新索引，极大提升写入性能
• 对于不会进行修改的表，支持压缩表，极大减少磁盘空间占用
• 创建一张表，对会应三个文件, *.frm 记录表结构, *.myd 数据, *.myi 索引文件

InnoDB：

InnoDB在MySQL 5.5后成为默认索引，它的特点是：

• 支持行锁，采用MVCC来支持高并发
• 支持事务
• 支持外键
• 支持崩溃后的安全恢复
• 不支持全文索引
• 创建一张表，对会应一个文件 *.frm，数据存放到 ibdata1

总体来讲，MyISAM适合SELECT密集型的表，而InnoDB适合INSERT和UPDATE密集型的表

其他注意事项：

1、AND 型查询要点 (排除越多的条件放在前面) ：假设要查询满足条件 A，B 和 C 的文档，满足 A 的文档有4万，满足 B 的有 9K，满足 C 的是 200，那么应该用 C and B and A 这样只需要查询200条记录。

2、OR型查询要点（符合越多的条件放在前面）：OR型查询与AND查询恰好相反，匹配最多的查询语句放在最前面。

3、查询数据不建议使用 select * from table ，应用具体的字段列表代替“*”，不要返回用不到的无关字段，尤其是大数据列。

4、在分页查询中使用 limit关键字时，应重复考虑使用索引字段来筛选来避免全表扫描，如：

select c1, c2, c3 from table order by id asc limit 100, 100
1

应尽量配合where条件来使用(大数据量情况查询效率提升10倍)：

select c1, c2, c3 from table where id > 100 order by id asc limit 0, 100
1

5、在新建临时表时，如果一次性插入数据量很大，那么可以使用 select into 代替 create table，避免造成大量 log ，以提高速度；如果数据量不大，为了缓和系统表的资源，应先 create table，然后再insert。

6、如果使用到了临时表，在存储过程的最后务必将所有的临时表显式删除，先 truncate table ，然后 drop table ，这样可以避免系统表的较长时间锁定。但是，避免频繁创建和删除临时表，以减少系统表资源的消耗。

7、尽量避免使用游标，因为游标的效率较差，如果游标操作的数据超过1万行，那么就应该考虑改写。因此，使用基于游标的方法或临时表方法之前，应先寻找基于集的解决方案来解决问题，基于集的方法通常更有效。

8、与临时表一样，游标并不是不可使用。对小型数据集使用 FAST_FORWARD 游标通常要优于其他逐行处理方法，尤其是在必须引用几个表才能获得所需的数据时。在结果集中包括“合计”的例程通常要比使用游标执行的速度快。如果开发时间允许，基于游标的方法和基于集的方法都可以尝试一下，看哪一种方法的效果更好。

9、在所有的存储过程和触发器的开始处设置 SET NOCOUNT ON ，在结束时设置 SET NOCOUNT OFF 。无需在执行存储过程和触发器的每个语句后向客户端发送 DONE_IN_PROC 消息。

10、为提高系统并发能力，应尽量避免大事务操作，尽量避免向客户端返回大数据量，若数据量过大，应该考虑相应需求是否合理。

参数调优：

• wait_timeout：
  数据库连接闲置时间（长连接），闲置连接会占用内存资源。可以从默认的8小时减到半小时。

• max_user_connection：
  最大连接数，默认为0（无上限），最好设一个合理上限。

• thread_concurrency：
  并发线程数，设为CPU核数的两倍。

• key_buffer_size：
  索引块的缓存大小，增加会提升索引处理速度，对 MyISAM 表性能影响最大。对于内存 4G 左右，可设为 256M 或 384M，通过查询 show status like ‘key_read%’，保证 key_reads / key_read_requests 在0.1%以下最好

• innodb_buffer_pool_size ：
  缓存数据块和索引块，对InnoDB表性能影响最大。通过查询 show status like 'Innodb_buffer_pool_read%'，
  保证 (Innodb_buffer_pool_read_requests – Innodb_buffer_pool_reads) / Innodb_buffer_pool_read_requests 越高越好

• read_buffer_size：
  MySql 读入缓冲区大小。对表进行顺序扫描的请求将分配一个读入缓冲区，MySql会为它分配一段内存缓冲区。如果对表的顺序扫描请求非常频繁，可以通过增加该变量值以及内存缓冲区大小提高其性能

• sort_buffer_size
  MySql 执行排序使用的缓冲大小。如果想要增加 ORDER BY 的速度，首先看是否可以让 MySQL 使用索引而不是额外的排序阶段。如果不能，可以尝试增加 sort_buffer_size 变量的大小

• 其他参数

D、读写分离

也是目前常用的优化，从库读主库写，一般不要采用双主或多主引入很多复杂性，尽量采用文中的其他方案来提高性能。同时目前很多拆分的解决方案同时也兼顾考虑了读写分离。

• 分库分表
• 水平拆分
• 垂直拆分
• 升级硬件

根据 MySQL 是 CPU 密集型还是I/O密集型，通过提升 CPU 和内存、使用 SSD，都能显著提升 MySQL性能。

E、缓存应用

MySQL内部：

在系统调优参数介绍了相关设置

数据访问层：

比如 MyBatis 针对 SQL 语句做缓存，而 Hibernate 可以精确到单个记录，这里缓存的对象主要是持久化对象 Persistence Object

应用服务层：

这里可以通过编程手段对缓存做到更精准的控制和更多的实现策略，这里缓存的对象是数据传输对象 Data Transfer Object

Web层：

针对web页面做缓存

F、用户端的缓存

可以根据实际情况在一个层次或多个层次结合加入缓存。这里重点介绍下服务层的缓存实现，目前主要有两种方式：

1、直写式 (Write Through) ：在数据写入数据库后，同时更新缓存，维持数据库与缓存的一致性。这也是当前大多数应用缓存框架如 Spring Cache 的工作方式。这种实现非常简单，同步好，但效率一般。

2、回写式 (Write Back) ：当有数据要写入数据库时，只会更新缓存，然后异步批量的将缓存数据同步到数据库上。这种实现比较复杂，需要较多的应用逻辑，同时可能会产生数据库与缓存的不同步，但效率非常高。

二、案例

实际案例：

比如针对 1000w 条老数据加密处理

解决的思路：

查询优化 + 内存调优 + 分批处理 + 高效数据源 + 多线程 (线程池) + 反复调试

考虑客观因素：

硬件设施、网络宽带

假设1000w，分配20个线程，每页1万条，则分页总数：1000 / 1 = 1000页

线程平均处理分页数：(int)Math.floor(1000 / 20) = 50页

线程处理最大分页数：50页 + 1000 % 20 = 50页

则第一个线程处理 1 - 20页，

第二个线程处理 21- 40页，

第三个线程处理 41- 60页，

…

依次内推，第三十个线程，处理981 - 1000页

代码片段：

    /**
     * maysql 多线程 + 连接池，分页查询 + 批量更新示例
     * 注意：limit 参数不支持使用占位符?
     *
     * 10个线程，60万条数据，每页1w条，进行批量更新总耗时：53558毫秒，约0.90分钟（Where条件 + Limit N ）
     * 20个线程，60万条数据，每页1w条，进行批量更新总耗时：53558毫秒，约0.89分钟（Where条件 + Limit N ）
     * 20个线程，200万条数据，每页1w条，进行批量更新总耗时：160132毫秒，约2.67分钟（Where条件 + Limit N ）
     * 40个线程，200万条数据，每页1w条，进行批量更新总耗时：199220毫秒，约3.32分钟（Where条件 + Limit N ）
     *
     * 截止4月27日finance_ant_loan数据库数据：
     * LendingDetail, 总记录数 8729
     * LoanDetail, 总记录数 550417
     *
     * 截止4月27日finance_jd_loan数据库数据：
     * CUS， 总记录数 1006179
     * Loan，总记录数 32395990
     */
    @Test
    public void testMultiThreadBatchUpdate2(){
        Long beginTime = System.currentTimeMillis();
        int pageSize = 10000;
        int threads = 20;
        int resultCount = findCount();
 
        if (resultCount <= 0){
            log.info("未找到符合条件的记录！");
            return;
        }
 
        int pageTotal = (resultCount % pageSize == 0) ? (resultCount / pageSize) :  ((int)Math.floor(resultCount / pageSize) + 1);
        log.info("查询出数据库总计录数：{}", resultCount);
        log.info("每页数量：{}", pageSize);
        log.info("分页总数：{}", pageTotal);
 
        threads = getActualThreads(pageTotal, threads);
        int avgPages = (int)Math.floor(pageTotal / threads);
        int restPages = pageTotal % threads;
        log.info("子线程数：{}", threads);
        log.info("子线程平均处理分页数：{}", avgPages);
        log.info("子线程处理最大分页数：{}", avgPages + restPages);
 
        MultiThreadHandler handler = new MultiParallelThreadHandler();
        for (int i = 0; i < threads; i++){
            int fromPage = i * avgPages + 1;
            int endPage = i * avgPages + avgPages;
 
            if (i == threads - 1) {
                endPage = endPage + restPages;
            }
 
            String threadName = "thread" + (i + 1);
            log.info("Query child thread:{} process paging interval: [{}, {}]", threadName, fromPage, endPage);
            handler.addTask(new TestThread2(fromPage, endPage, pageSize));
        }
 
        try {
            handler.run();
        } catch (ChildThreadException e) {
            log.error(e.getAllStackTraceMessage());
        }
 
        log.info("【分页查询+批量更新】结束，受影响总记录数：{}，总耗时：{}毫秒", resultCount, System.currentTimeMillis() - beginTime);
    }
 
    private int findCount(){
        Connection connection = AntConnPool.getConnection();
        PreparedStatement preparedStatement;
 
        try {
            preparedStatement = connection.prepareStatement("SELECT COUNT(1) FROM test2 WHERE 1 = 1");
            ResultSet resultSet = preparedStatement.executeQuery();
            if (resultSet.next()){
                return resultSet.getInt(1);
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("数据库操作异常！！！");
        }finally {
            ConnMgr.closeConn(connection);
        }
        return 0;
    }
 
    /**
     * @description 获得实际线程数
     * @author Zack
     * @date 15:28 2018/5/7
     * @param pageTotal
     * @param threads
     * @return int
     */
    private int getActualThreads(final int pageTotal, final int threads){
        if ((int)Math.floor(pageTotal / threads) < 1 && threads > 1){
            return getActualThreads(pageTotal, threads - 1);
        }
        return threads;
    }

/**
 * @description 多线程处理分页
 * @author Zack
 * @date 2018/4/27
 * @version 1.0
 */
@Slf4j
public class TestThread2 extends MysqlParallelThread{
 
    public TestThread2(int fromPage, int endPage, int pageSize){
        super(fromPage, endPage, pageSize);
    }
 
    @Override
    public void run() {
        log.info("Query child thread:{} process paging interval: [{}, {}] started.", Thread.currentThread().getName(), getFromPage(), getEndPage());
        Long beginTime = System.currentTimeMillis();
        int maxId = 0;
        int fromIndex = (getFromPage() - 1) * getPageSize();
        Connection connection = AntConnPool.getConnection();
 
        try{
            for (int pageNo = getFromPage(); pageNo <= getEndPage(); pageNo++){
                if (maxId != 0){
                    fromIndex = 0;
                }
                maxId = batchUpdate(findList(connection, fromIndex, maxId), pageNo);
            }
        }catch (Exception e){
            throw new RuntimeException(Thread.currentThread().getName() + ": throw exception");
        }finally {
            ConnMgr.closeConn(connection);
            log.info("Query child thread:{} process paging interval: [{}, {}] end. cost：{} ms.", Thread.currentThread().getName(), getFromPage(), getEndPage(), (System.currentTimeMillis() - beginTime));
        }
    }
 
    private ResultSet findList(Connection connection, int fromIndex, int maxId){
        try {
            PreparedStatement preparedStatement = connection.prepareStatement("SELECT id, name, total FROM test2 WHERE id > ? ORDER BY id ASC limit "+fromIndex+", " + getPageSize(), ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            preparedStatement.setInt(1, maxId);
            return preparedStatement.executeQuery();
        }catch (Exception e){
            e.printStackTrace();
            log.error("数据库操作异常！！！");
        }
        return null;
    }
 
    private int batchUpdate(ResultSet resultSet, int pageNo) {
        Long beginTime = System.currentTimeMillis();
        Connection connection = AntConnPool.getConnection();
        PreparedStatement preparedStatement;
 
        try {
            connection.setAutoCommit(false);
            preparedStatement = connection.prepareStatement("UPDATE test2 SET name = ?, total = ? WHERE id = ?");
 
            if (Objects.isNull(resultSet) || resultSet.wasNull()){
                log.info("查询第{}页数据为空！", pageNo);
                return 0;
            }
 
            while (resultSet.next()){
                // 加密处理
                preparedStatement.setString(1, SecUtil.encryption(resultSet.getString(2)));
                preparedStatement.setInt(2, resultSet.getInt(3) + 1);
                preparedStatement.setInt(3, resultSet.getInt(1));
                preparedStatement.addBatch();
            }
 
            int[] countArray = preparedStatement.executeBatch();
            connection.commit();
 
            // 游标移至最后一行
            resultSet.last();
            int maxId = resultSet.getInt(1);
            log.info("子线程{}批量更新MYSQL第{}页结束，maxId：{}, 受影响记录数：{}，耗时：{}毫秒", Thread.currentThread().getName(), pageNo, maxId, countArray.length, System.currentTimeMillis() - beginTime);
            return maxId;
        }catch (Exception e){
            e.printStackTrace();
            log.error("数据库操作异常！！！");
        }finally {
            ConnMgr.closeConn(connection);
        }
        return 0;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185

总结

使用多线程合理分配执行任务，避免数据重复执行和漏执行。

---

原文链接：https://blog.csdn.net/chivydrs/article/details/81670475