MYSQL开发性能研究之批量插入数据的优化方法

一、我们遇到了什么问题

在标准sql里面，我们通常会写下如下的sql insert语句。

insert into tbl_test (id) values(1);

很显然,在mysql中，这样的方式也是可行的。但是当我们需要批量插入数据的时候，这样的语句却会出现性能问题。例如说，如果有需要插入100000条数据，那么就需要有100000条insert语句，每一句都需要提交到关系引擎那里去解析，优化，然后才能够到达存储引擎做真的插入工作。

正是由于性能的瓶颈问题，mysql官方文档也就提到了使用批量化插入的方式，也就是在一句insert语句里面插入多个值。即，

insert into tbl_test (id) values (1), (2), (3)

这样的做法确实也可以起到加速批量插入的功效，原因也不难理解，由于提交到服务器的insert语句少了，网络负载少了，最主要的是解析和优化的时间看似增多，但是实际上作用的数据行却实打实地多了。所以整体性能得以提高。根据网上的一些说法，这种方法可以提高几十倍。

然而，我在网上也看到过另外的几种方法，比如说预处理sql，比如说批量提交。那么这些方法的性能到底如何？本文就会对这些方法做一个比较。

二、比较环境和方法
我的环境比较苦逼，基本上就是一个落后的虚拟机。只有2核，内存为6g。操作系统是susi linux，mysql版本是5.6.15。

可以想见，这个机子的性能导致了我的tps一定非常低，所以下面的所有数据都是没有意义的，但是趋势却不同，它可以看出整个插入的性能走向。

由于业务特点，我们所使用的表非常大，共有195个字段，且写满（每个字段全部填满，包括varchar）大致会有略小于4kb的大小，而通常来说，一条记录的大小也有3kb。

由于根据我们的实际经验，我们很肯定的是，通过在一个事务中提交大量insert语句可以大幅度提高性能。所以下面的所有测试都是建立在每插入5000条记录提交一次的做法之上。

最后需要说明的是，下面所有的测试都是通过使用mysql c api进行的，并且使用的是innodb存储引擎。

三、比较方法

理想型测试（一）——方法比较

目的：找出理想情况下最合适的插入机制

关键方法：

1. 每个进/线程按主键顺序插入

2. 比较不同的插入方法

3. 比较不同进/线程数量对插入的影响

*“普通方法”指的是一句insert只插入一个value的情况。

*“预处理sql”指的是使用预处理mysql c api的情况。

* “多表值sql(10条）”是使用一句insert语句插入10条记录的情况。为什么是10条？后面的验证告诉了我们这样做性能最高。

结论，很显然，从三种方法的趋势上来看，多表值sql(10条）的方式最为高效。

理想型测试（二）——多表值sql条数比较

很显然，在数据量提高的情况下，每条insert语句插入10条记录的做法最为高效。

理想型测试（三）——连接数比较

结论：在2倍与cpu核数的连接和操作的时候，性能最高

一般性测试—— 根据我们的业务量进行测试

目的：最佳插入机制适合普通交易情况？

关键方法：

1. 模拟生产数据(每条记录约3kb)

2. 每个线程主键乱序插入

很显然，如果是根据主键乱序插入的话，性能会有直线下降的情况。这一点其实和innodb的内部实现原理所展现出来的现象一致。但是仍然可以肯定的是，多表值sql(10条）的情况是最佳的。

压力测试

目的：最佳插入机制适合极端交易情况？

关键方法：

1. 将数据行的每一个字段填满（每条记录约为4kb）

2. 每个线程主键乱序插入

结果和我们之前的规律类似，性能出现了极端下降。并且这里验证了随着记录的增大（可能已经超过了一个page的大小，毕竟还有slot和page head信息占据空间），会有page split等现象，性能会下降。

四、结论

根据上面的测试，以及我们对innodb的了解，我们可以得到如下的结论。

•采用顺序主键策略（例如自增主键，或者修改业务逻辑，让插入的记录尽可能顺序主键）

•采用多值表（10条）插入方式最为合适

•将进程/线程数控制在2倍cpu数目相对合适

五、附录

我发现网上很少有完整的针对mysql 预处理sql语句的例子。这里给出一个简单的例子。

--建表语句
create table tbl_test 
(
  pri_key varchar(30), 
  nor_char char(30), 
  max_num decimal(8,0), 
  long_num decimal(12, 0), 
  rec_upd_ts timestamp
);

c代码

#include <string.h>
#include <iostream>
#include <mysql.h>
#include <sys/time.h>
#include <sstream>
#include <vector>
 
using namespace std;
 
#define string_len 30
  
char    pri_key            [string_len]= "123456"; 
char    nor_char           [string_len]= "abcabc"; 
char    rec_upd_ts          [string_len]= "now()"; 
 
bool subtimeval(timeval &result, timeval &begin, timeval &end)
{
  if ( begin.tv_sec>end.tv_sec ) return false;
 
  if ( (begin.tv_sec == end.tv_sec) && (begin.tv_usec > end.tv_usec) )  
    return  false;
 
  result.tv_sec = ( end.tv_sec - begin.tv_sec );  
  result.tv_usec = ( end.tv_usec - begin.tv_usec );  
 
  if (result.tv_usec<0) {
    result.tv_sec--;
    result.tv_usec+=1000000;} 
  return true;
}
 
int main(int argc, char ** argv)
{
  int32 ret = 0;
  char errmsg[200] = {0};
  int sqlcode = 0;
 
  timeval tbegin, tend, tdiff;
   
  const char* precompile_statment2 = "insert into `tbl_test`( pri_key, nor_char, max_num, long_num, rec_upd_ts) values(?, ?, ?, ?, ?)";
   
  mysql conn;
  mysql_init(&conn);
   
  if (mysql_real_connect(&conn, "127.0.0.1", "dba", "abcdefg", "testdb", 3306, null, 0) == null)
  {
    fprintf(stderr, " mysql_real_connect, 2 failed\n");
    exit(0);
  }
   
  mysql_stmt  *stmt = mysql_stmt_init(&conn);
  if (!stmt)
  {
   fprintf(stderr, " mysql_stmt_init, 2 failed\n");
   fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
   exit(0);
  }
   
  if (mysql_stmt_prepare(stmt, precompile_statment2, strlen(precompile_statment2)))
  {
   fprintf(stderr, " mysql_stmt_prepare, 2 failed\n");
   fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
   exit(0);
  }
   
  int i = 0; 
  int max_num = 3;
  const int field_num = 5;
  while (i < max_num)
  {
    //mysql_bind  bind[196] = {0};
    mysql_bind  bind[field_num];
    memset(bind, 0, field_num * sizeof(mysql_bind));
   
    unsigned long str_length = strlen(pri_key);
    bind[0].buffer_type  = mysql_type_string;
    bind[0].buffer    = (char *)pri_key;
    bind[0].buffer_length = string_len;
    bind[0].is_null    = 0;
    bind[0].length    = &str_length;
     
    unsigned long str_length_nor = strlen(nor_char);
    bind[1].buffer_type  = mysql_type_string;
    bind[1].buffer    = (char *)nor_char;
    bind[1].buffer_length = string_len;
    bind[1].is_null    = 0;
    bind[1].length    = &str_length_nor;
     
    bind[2].buffer_type  = mysql_type_long;
    bind[2].buffer    = (char*)&max_num;
    bind[2].is_null    = 0;
    bind[2].length    = 0;
     
    bind[3].buffer_type  = mysql_type_long;
    bind[3].buffer    = (char*)&max_num;
    bind[3].is_null    = 0;
    bind[3].length    = 0;
     
    mysql_time ts;
    ts.year= 2002;
    ts.month= 02;
    ts.day= 03;
    ts.hour= 10;
    ts.minute= 45;
    ts.second= 20;
     
    unsigned long str_length_time = strlen(rec_upd_ts);
    bind[4].buffer_type  = mysql_type_timestamp;
    bind[4].buffer    = (char *)&ts;
    bind[4].is_null    = 0;
    bind[4].length    = 0;
     
    if (mysql_stmt_bind_param(stmt, bind))
    {
      fprintf(stderr, " mysql_stmt_bind_param, 2 failed\n");
      fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
      exit(0);
    }
     
    cout << "before execute\n";
    if (mysql_stmt_execute(stmt))
    {
     fprintf(stderr, " mysql_stmt_execute, 2 failed\n");
     fprintf(stderr, " %s\n", mysql_stmt_error(stmt));
     exit(0);
    }
    cout << "after execute\n";
     
    i++;
  }
   
  mysql_commit(&conn);
   
  mysql_stmt_close(stmt);
 
  return 0;  
}

以上就是mysql批量插入数据的优化方法,建议大家也可以多看下以前的文章。