breeze/workflow
1
0
Fork 0
mirror of https://github.com/sogou/workflow.git synced 2026-02-08 01:33:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update tutorial-12-mysql_cli.md

Browse Source
This commit is contained in:
liyingxin
2021-05-19 21:51:44 +08:00
committed by GitHub
parent 570fec8458
commit c37d0d78c6
1 changed files with 73 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
docs/tutorial-12-mysql_cli.md
View File

@@ -21,7 +21,7 @@ mysql://username:password@host:port/dbname?character_set=charset&character_set_r
- dbname为要用的数据库名，一般如果SQL语句只操作一个db的话建议填写；
- 如果用户在这一层有upstream选取需求，可以参考[upstream文档](../docs/about-upstream.md)；
- 如果用户在这一层有upstream选取需求，可以参考[upstream文档](/docs/about-upstream.md)；
- character_set为client的字符集，等价于使用官方客户端启动时的参数``--default-character-set``的配置，默认utf8，具体可以参考MySQL官方文档[character-set.html](https://dev.mysql.com/doc/internals/en/character-set.html)。
@@ -68,7 +68,9 @@ int main(int argc, char *argv[])
因为我们的交互命令中不支持选库（**USE**命令），所以，如果SQL语句中有涉及到**跨库**的操作，则可以通过**db_name.table_name**的方式指定具体哪个库的哪张表。
其他所有命令都可以**拼接**到一起通过 ``set_query()`` 传给WFMySQLTask（包括INSERT/UPDATE/SELECT/PREPARE/CALL）。
其他所有命令都可以**拼接**到一起通过 ``set_query()`` 传给WFMySQLTask（包括INSERT/UPDATE/SELECT/DELETE/PREPARE/CALL）。
拼接的命令会被按序执行直到命令发生错误，前面的命令会执行成功。
举个例子：
~~~cpp
@@ -77,52 +79,72 @@ req->set_query("SELECT * FROM table1; CALL procedure1(); INSERT INTO table3 (id)
# 结果解析
与workflow其他任务类似，可以用task->get_resp()拿到**MySQLResponse**，我们可以通过**MySQLResultCursor**遍历结果集及其中的每个列的信息**MySQLField**、每行和每个**MySQLCell**。具体接口可以查看：[MySQLResult.h](../src/protocol/MySQLResult.h)
与workflow其他任务类似，可以用``task->get_resp()``拿到**MySQLResponse**，我们可以通过**MySQLResultCursor**遍历结果集。具体接口可以查看：[MySQLResult.h](/src/protocol/MySQLResult.h)
一次请求所对应的回复中，其数据是一个三维结构：
- 一个回复中包含了一个或多个结果集（result set）；
- 一个结果集包含了一行或多行（row）；
- 一行包含了一列或多个列，或者说一到多个阈（Field/Cell）；
- 一个结果集的类型可能是**MYSQL_STATUS_GET_RESULT**或者**MYSQL_STATUS_OK**；
- **MYSQL_STATUS_GET_RESULT**类型的结果集包含了一行或多行（row）；
- 一行包含了一列或多个列，或者说一到多个阈（Field/Cell），具体数据结构为**MySQLField**和**MySQLCell**；
结果集的两种类型，可以通过``cursor->get_cursor_status()``进行判断：
| |MYSQL_STATUS_GET_RESULT|MYSQL_STATUS_OK|
|------|-----------------------|---------------|
|SQL命令|SELECT（包括存储过程中的每一个SELECT）|INSERT / UPDATE / DELETE / ...|
|对应语义|读操作，一个结果集表示一份读操作返回的二维表|写操作，一个结果集表示一个写操作是否成功|
|主要接口|fetch_fields();</br>fetch_row(&row_arr);</br>...|get_insert_id();</br>get_affected_rows();</br>...|
由于拼接语句可能存在错误，因此这种情况，可以通过**MySQLResultCursor**拿到前面正确执行过的语句多个结果集，以及最后判断``resp->get_packet_type()``为**MYSQL_PACKET_ERROR**时，通过``resp->get_error_code()``和``resp->get_error_msg()``拿到具体错误信息。
一个包含n条**SELECT**语句的**存储过程**，会返回n个**MYSQL_STATUS_GET_RESULT**的结果集和1个**MYSQL_STATUS_OK**的结果集，用户自行忽略此**MYSQL_STATUS_OK**结果集即可。
具体使用从外到内的步骤应该是：
1. 判断任务状态（代表通信层面状态）：用户通过判断 **task->get_state()** 等于WFT_STATE_SUCCESS来查看任务执行是否成功；
1. 判断任务状态（代表通信层面状态）：用户通过判断 ``task->get_state()`` 等于**WFT_STATE_SUCCESS**来查看任务执行是否成功；
2. 判断回复包类型（代表返回包解析状态）：调用 **resp->get_packet_type()** 查看最后一条MySQL语句的返回包类型，常见的几个类型为：
- MYSQL_PACKET_OK：返回非结果集的请求: 解析成功；
- MYSQL_PACKET_EOF：返回结果集的请求: 解析成功；
- MYSQL_PACKET_ERROR：请求:失败；
- MYSQL_PACKET_OK：成功，可以用cursor遍历结果；
- MYSQL_PACKET_EOF：成功，可以用cursor遍历结果；
- MYSQL_PACKET_ERROR：失败或部分失败，成功的部分可以用cursor遍历结果；
3. 判断结果集状态（代表结果集读取状态）：用户可以使用MySQLResultCursor读取结果集中的内容，因为MySQL server返回的数据是多结果集的，因此一开始cursor会**自动指向第一个结果集**的读取位置。通过 **cursor->get_cursor_status()** 可以拿到的几种状态：
- MYSQL_STATUS_GET_RESULT：有数据可读；
- MYSQL_STATUS_END：当前结果集已读完最后一行；
- MYSQL_STATUS_EOF：所有结果集已取完；
- MYSQL_STATUS_OK：此回复包为非结果集包，无需通过结果集接口读数据；
3. 遍历结果集。用户可以使用**MySQLResultCursor**读取结果集中的内容，因为MySQL server返回的数据是多结果集的，因此一开始cursor会**自动指向第一个结果集**的读取位置。
4. 判断结果集状态（代表结果集读取状态）：通过 ``cursor->get_cursor_status()`` 可以拿到的几种状态：
- MYSQL_STATUS_GET_RESULT：此结果集为读请求类型；
- MYSQL_STATUS_END：读结果集已读完最后一行；
- MYSQL_STATUS_OK：此结果集为写请求类型；
- MYSQL_STATUS_ERROR：解析错误；
4. 读取columns中每个field：
5. 读取**MYSQL_STATUS_OK**结果集中的基本内容：
- ``unsigned long long get_affected_rows() const;``
- ``unsigned long long get_insert_id() const;``
- ``int get_warnings() const;``
- ``std::string get_info() const;``
6. 读取**MYSQL_STATUS_GET_RESULT**结果集中的columns中每个field：
- ``int get_field_count() const;``
- ``const MySQLField *fetch_field();``
- ``const MySQLField *const *fetch_fields() const;``
5. 读取每一行：按行读取可以使用 **cursor->fetch_row()** 直到返回值为false。其中会移动cursor内部对当前结果集的指向每行的offset：
7. 读取**MYSQL_STATUS_GET_RESULT**结果集中的每一行：按行读取可以使用 ``cursor->fetch_row()`` 直到返回值为false。其中会移动cursor内部对当前结果集的指向每行的offset：
- ``int get_rows_count() const;``
- ``bool fetch_row(std::vector<MySQLCell>& row_arr);``
- ``bool fetch_row(std::map<std::string, MySQLCell>& row_map);``
- ``bool fetch_row(std::unordered_map<std::string, MySQLCell>& row_map);``
- ``bool fetch_row_nocopy(const void **data, size_t *len, int *data_type);``
6. 直接把当前结果集的所有行拿出：所有行的读取可以使用 **cursor->fetch_all()** ，内部用来记录行的cursor会直接移动到最后；cursor状态会变成MYSQL_STATUS_END：
8. 直接把当前**MYSQL_STATUS_GET_RESULT**结果集的所有行拿出：所有行的读取可以使用 **cursor->fetch_all()** ，内部用来记录行的cursor会直接移动到最后；当前cursor状态会变成**MYSQL_STATUS_END**：
- ``bool fetch_all(std::vector<std::vector<MySQLCell>>& rows);``
7. 返回当前结果集的头部：如果有必要重读这个结果集，可以使用 **cursor->rewind()** 回到当前结果集头部，再通过第5步或第6步进行读取；
9. 返回当前**MYSQL_STATUS_GET_RESULT**结果集的头部：如果有必要重读这个结果集，可以使用 **cursor->rewind()** 回到当前结果集头部，再通过第7步或第8步进行读取；
8. 拿到下一个结果集：因为MySQL server返回的数据包可能是包含多结果集的（比如每个select语句为一个结果集；或者call procedure返回的多结果集数据），因此用户可以通过 **cursor->next_result_set()** 跳到下一个结果集，返回值为false表示所有结果集已取完。
10. 拿到下一个结果集：因为MySQL server返回的数据包可能是包含多结果集的（比如每个select/insert/...语句为一个结果集；或者call procedure返回的多结果集数据），因此用户可以通过 **cursor->next_result_set()** 跳到下一个结果集，返回值为false表示所有结果集已取完。
9. 返回第一个结果集：**cursor->first_result_set()** 可以让我们返回到所有结果集的头部，然后可以从第3步开始重新拿数据；
11. 返回第一个结果集：**cursor->first_result_set()** 可以让我们返回到所有结果集的头部，然后可以从第4步开始重新拿数据；
10. 每列具体数据MySQLCell：第5步中读取到的一行，由多列组成，每列结果为MySQLCell，基本使用接口有：
- ``int get_data_type();`` 返回MYSQL_TYPE_LONG、MYSQL_TYPE_STRING...具体参考[mysql_types.h](../src/protocol/mysql_types.h)
12. **MYSQL_STATUS_GET_RESULT**结果集每列具体数据MySQLCell：第7步中读取到的一行，由多列组成，每列结果为MySQLCell，基本使用接口有：
- ``int get_data_type();`` 返回MYSQL_TYPE_LONG、MYSQL_TYPE_STRING...具体参考[mysql_types.h](/src/protocol/mysql_types.h)
- ``bool is_TYPE() const;`` TYPE为int、string、ulonglong，判断是否是某种类型
- ``TYPE as_TYPE() const;`` 同上，以某种类型读出MySQLCell的数据
- ``void get_cell_nocopy(const void **data, size_t *len, int *data_type) const;`` nocopy接口
@@ -142,15 +164,30 @@ void task_callback(WFMySQLTask *task)
bool test_first_result_set_flag = false;
bool test_rewind_flag = false;
begin:
// step-3. 判断结果集状态
if (cursor.get_cursor_status() == MYSQL_STATUS_GET_RESULT)
// step-2. 判断回复包其他状态
if (resp->get_packet_type() == MYSQL_PACKET_ERROR)
{
do {
fprintf(stderr, "cursor_status=%d field_count=%u rows_count=%u ",
cursor.get_cursor_status(), cursor.get_field_count(), cursor.get_rows_count());
fprintf(stderr, "ERROR. error_code=%d %s\n",
task->get_resp()->get_error_code(),
task->get_resp()->get_error_msg().c_str());
}
// step-4. 读取每个fields。这是个nocopy api
begin:
// step-3. 遍历结果集
do {
// step-4. 判断结果集状态
if (cursor.get_cursor_status() == MYSQL_STATUS_OK)
{
// step-5. MYSQL_STATUS_OK结果集的基本内容
fprintf(stderr, "OK. %llu rows affected. %d warnings. insert_id=%llu.\n",
cursor.get_affected_rows(), cursor.get_warnings(), cursor.get_insert_id());
}
eles if (cursor.get_cursor_status() == MYSQL_STATUS_GET_RESULT)
{
fprintf(stderr, "field_count=%u rows_count=%u ",
cursor.get_field_count(), cursor.get_rows_count());
// step-6. 读取每个fields。这是个nocopy api
const MySQLField *const *fields = cursor.fetch_fields();
for (int i = 0; i < cursor.get_field_count(); i++)
{
@@ -159,18 +196,18 @@ begin:
fields[i]->get_name().c_str(), datatype2str(fields[i]->get_data_type()));
}
// step-6. 把所有行读出，也可以while (cursor.fetch_row(map/vector)) 按step-5拿每一行
// step-8. 把所有行读出，也可以while (cursor.fetch_row(map/vector)) 按step-7拿每一行
std::vector<std::vector<MySQLCell>> rows;
cursor.fetch_all(rows);
for (unsigned int j = 0; j < rows.size(); j++)
{
// step-10. 具体每个cell的读取
// step-12. 具体每个cell的读取
for (unsigned int i = 0; i < rows[j].size(); i++)
{
fprintf(stderr, "[%s][%s]", fields[i]->get_name().c_str(),
datatype2str(rows[j][i].get_data_type()));
// step-10. 判断具体类型is_string()和转换具体类型as_string()
// step-12. 判断具体类型is_string()和转换具体类型as_string()
if (rows[j][i].is_string())
{
std::string res = rows[j][i].as_string();
@@ -180,13 +217,13 @@ begin:
} // else if ...
}
}
// step-8. 拿下一个结果集
// step-10. 拿下一个结果集
} while (cursor.next_result_set());
if (test_first_result_set_flag == false)
{
test_first_result_set_flag = true;
// step-9. 返回第一个结果集
// step-11. 返回第一个结果集
cursor.first_result_set();
goto begin;
}
@@ -194,21 +231,11 @@ begin:
if (test_rewind_flag == false)
{
test_rewind_flag = true;
// step-7. 返回当前结果集头部
// step-9. 返回当前结果集头部
cursor.rewind();
goto begin;
}
}
// step-2. 判断回复包其他状态
else if (resp->get_packet_type() == MYSQL_PACKET_OK)
{
fprintf(stderr, "OK. %llu rows affected. %d warnings. insert_id=%llu.\n",
task->get_resp()->get_affected_rows(),
task->get_resp()->get_warnings(),
task->get_resp()->get_last_insert_id());
}
else
fprintf(stderr, "Abnormal packet_type=%d\n", resp->get_packet_type());
return;
}
@@ -216,7 +243,7 @@ begin:
# WFMySQLConnection
由于我们是高并发异步客户端，这意味着我们对一个server的连接可能会不止一个。而MySQL的事务和预处理都是带状态的，为了保证一次事务或预处理独占一个连接，用户可以使用我们封装的二级工厂WFMySQLConnection来创建任务，每个WFMySQLConnection保证独占一个连接，具体参考[WFMySQLConnection.h](../src/client/WFMySQLConnection.h)。
由于我们是高并发异步客户端，这意味着我们对一个server的连接可能会不止一个。而MySQL的事务和预处理都是带状态的，为了保证一次事务或预处理独占一个连接，用户可以使用我们封装的二级工厂WFMySQLConnection来创建任务，每个WFMySQLConnection保证独占一个连接，具体参考[WFMySQLConnection.h](/src/client/WFMySQLConnection.h)。
### 1. WFMySQLConnection的创建与初始化