mysql中，当使用的数据类型是longblob时，如何实现流式的读写，即每次读出或写入一块数据，分多次操作。
一般的游标函数明显是无法完成这个需求的，从mysql.h中看其实现就明白了：
//典型的查询的代码如下
int Query(const char* sql, int len, vector< vector<string> >& out)
{
if (0!=mysql_real_query(&m_client, sql, len))
{
m_ErrorCode = (int)mysql_errno(&m_client);
ERR_LOG("%s", mysql_error(&m_client));
return -1;
}
MYSQL_RES* result = mysql_store_result(&m_client);
if (NULL==result)
{
m_ErrorCode = (int)mysql_errno(&m_client);
ERR_LOG("%s", mysql_error(&m_client));
return -2;
}
MYSQL_ROW row;
unsigned int i;
unsigned int num_fields = mysql_num_fields(result);
int row_count = 0;
while ((row = mysql_fetch_row(result)))
{
++row_count;
unsigned long* lengths = mysql_fetch_lengths(result);
vector<string> row_data;
for(i = 0; i<num_fields; ++i)
{
row_data.push_back(string(row[i], (int)lengths[i]));
}
out.push_back(row_data);
}
mysql_free_result(result);
return row_count;
}
MYSQL_ROW的定义为：
typedef char **MYSQL_ROW; /* return data as array of strings */
可见，执行函数后，所有数据都已经存在于本地的内存中了。
假设用这样的方法读取LONGBLOB中2G的字段，则一定要2G数据全部读到本地内存后才返回，或者在内存不足的时候抛出out of memory的错误。
mysql中对LONGBLOB字段实现流式的读写，必须要使用Prepared Statement系列的函数。
实现流式读写的关键函数是：
my_bool mysql_stmt_send_long_data(MYSQL_STMT *stmt, unsigned int parameter_number, const char *data, unsigned long length)
多次调用，每次写入一块数据
int mysql_stmt_fetch_column(MYSQL_STMT *stmt, MYSQL_BIND *bind, unsigned int column, unsigned long offset)
多次调用，每次从结果集的某个字段里读出一块数据
下面是分段写入和读取的例子：
//---------test_write_longblob_3.cpp-------------------------------------------
#include <stdio.h>
#include <mysql.h>
#include <assert.h>
#include <string.h>
#include <stdlib.h>
#define ERR_LOG(format, ...) printf("%s %d:" format "\n", __FILE__, __LINE__, ##__VA_ARGS__)
#define MYSQL_SET_BIND(bind_, type_, buffer_, is_null_, len_) \
{ \
bind_.buffer_type = type_; \
bind_.buffer = buffer_; \
bind_.is_null = is_null_; \
bind_.length = len_; \
}
/*
测试数据库
create table test_longbolb(v longblob);
*/
void test_write(MYSQL& client)
{
MYSQL_STMT* stmt = mysql_stmt_init(&client);
assert(NULL!=stmt);
const char* sql = "insert into test_longbolb (v) values(?)";
int sql_len = strlen(sql);
int ret = mysql_stmt_prepare(stmt, sql, sql_len);
assert(0==ret);
ERR_LOG("param count:%d", (int)mysql_stmt_param_count(stmt));
char null_flag = 0;
MYSQL_BIND param = {0};
MYSQL_SET_BIND(param, MYSQL_TYPE_LONG_BLOB, NULL, &null_flag, NULL);
ret = mysql_stmt_bind_param(stmt, ¶m);
assert(0==ret);
//
for (int i=0; i<10; ++i)
{
char buf[10];
int buf_len = snprintf(buf, sizeof(buf), "%d\n", i);
char ret1 = mysql_stmt_send_long_data(stmt, 0, buf, buf_len);
if (ret1!=0)
{
ERR_LOG("code=%d, msg=%s", (int)mysql_errno(&client), mysql_error(&client));
return;
}
}
//
ret = mysql_stmt_execute(stmt);
assert(0==ret);
mysql_stmt_close(stmt);
}
int main(int argc, char* argv[])
{
if (argc<6)
{
printf("usage:%s <host> <port> <user> <pass> <db>\n", argv[0]);
return 1;
}
MYSQL client;
mysql_init(&client);
if (NULL==mysql_real_connect(&client, argv[1], //host,
argv[3], //user,
argv[4], //pass,
argv[5], //db,
atoi(argv[2]), //port,
NULL, 0))
{
ERR_LOG("code=%d, msg=%s", (int)mysql_errno(&client), mysql_error(&client));
return 1;
}
test_write(client);
ERR_LOG("OK");
return 1;
}
/*
g++ -o test_write_longblob_3.o -c test_write_longblob_3.cpp -g -Wall -Werror
g++ -o test_write_longblob_3 test_write_longblob_3.o -lmysqlclient -static -pthread
./test_write_longblob_3 192.168.0.100 3306 root test123 test
*/
//---------end test_write_longblob_3.cpp---------------------------------------
下面是读取的例子：
//---------test_read_longblob_3.cpp--------------------------------------------
#include <stdio.h>
#include <mysql.h>
#include <assert.h>
#include <string.h>
#include <stdlib.h>
#define ERR_LOG(format, ...) printf("%s %d:" format "\n", __FILE__, __LINE__, ##__VA_ARGS__)
/*
测试数据库
create table test_longbolb(v longblob);
*/
void test_write(MYSQL& client)
{
MYSQL_STMT* stmt = mysql_stmt_init(&client);
assert(NULL!=stmt);
const char* sql = "select v from test_longbolb";
int sql_len = strlen(sql);
int ret = mysql_stmt_prepare(stmt, sql, sql_len);
assert(0==ret);
ERR_LOG("param count:%d", (int)mysql_stmt_param_count(stmt));
//
MYSQL_BIND result = {0};
unsigned long total_length = 0;
result.buffer_type = MYSQL_TYPE_LONG_BLOB;
result.length = &total_length;
ret = mysql_stmt_bind_result(stmt, &result);
assert(0==ret);
ret = mysql_stmt_execute(stmt);
assert(0==ret);
ret = mysql_stmt_store_result(stmt);
assert(0==ret);
//while (mysql_stmt_fetch(stmt)!=0)
for (;;)
{
ret = mysql_stmt_fetch(stmt);
if (ret!=0 && ret!=MYSQL_DATA_TRUNCATED) breal;
int start = 0;
char buf[1024] = {0};
printf("total_length=%lu\n", total_length);
while (start<(int)total_length)
{
result.buffer = (buf+start);
result.buffer_length = 3; //每次读这么长
ret = mysql_stmt_fetch_column(stmt, &result, 0, start);
if (ret!=0)
{
ERR_LOG("code=%d, msg=%s", (int)mysql_errno(&client), mysql_error(&client));
return;
}
start += result.buffer_length;
}
printf("%.*s\n", total_length, buf);
}
mysql_stmt_close(stmt);
}
int main(int argc, char* argv[])
{
if (argc<6)
{
printf("usage:%s <host> <port> <user> <pass> <db>\n", argv[0]);
return 1;
}
MYSQL client;
mysql_init(&client);
if (NULL==mysql_real_connect(&client, argv[1], //host,
argv[3], //user,
argv[4], //pass,
argv[5], //db,
atoi(argv[2]), //port,
NULL, 0))
{
ERR_LOG("code=%d, msg=%s", (int)mysql_errno(&client), mysql_error(&client));
return 1;
}
test_write(client);
ERR_LOG("OK");
return 1;
}
/*
g++ -o test_read_longblob_3.o -c test_read_longblob_3.cpp -g -Wall -Werror
g++ -o test_read_longblob_3 test_read_longblob_3.o -lmysqlclient -static -pthread
./test_read_longblob_3 192.168.0.100 3306 root test123 test
*/
//---------end test_read_longblob_3.cpp----------------------------------------