基于MySQL内核的SQL限流设计与实现

// 更新static bool update_delimiter(sys_var *self, THD *thd, enum_var_type type){  reload_rules(thd);}
// 校验static bool check_delimiter(sys_var *self, THD *thd, set_var *var){  judge(var->value);  judge(str);  judge(length);}

int mysqld_main(int argc, char **argv){  ...  load_rules();  ...}

void dec_conc(THD *thd, int command){  // 根据查询类型在对应链表找节点  node = find_by_id(list, thd->id);
  // 并发数量减1  if (node) {    __sync_sub_and_fetch(&(node->concur), 1);  }    // 重置状态  thd->id = 0;}

void clean_up(){  cleanup();}

/* The flow control does not take effect on system tables */  if (check_system_table(first_table->db)) {    return ret;  }

bool check_rule_matched(THD* thd, LIST* list){  while (满足条件，无异常) {    // 根据大小写开关是否打开，分别进行模式串匹配    it = find(query_str, item->key_array[nums]);          // 如果it为空，没有匹配到，查看下一个list，否则继续匹配当前限流规则节点    judge();  }
  // 匹配成功，或者对下一个节点进行匹配}

class Du_table_access {public:  Du_table_access() : m_drop_thd_object(NULL) {}  virtual ~Du_table_access() {}
  // 初始化打开表的环境、锁表并且打开表  bool init(THD **thd, TABLE **table, bool is_write);    // 关闭表，清理环境  bool deinit(THD *thd, TABLE *table, bool error, bool need_commit);    // 设置打开表的策略  void before_open(THD *thd);    // 如果需要的话创建线程，大部分时候并不需要，因为手动执行读取数据的时候已经在线程中了  THD *create_thd();    // 如果手动创建了 thd，则需要手动清理  void drop_thd(THD *thd);};

DELIMITER $$
CREATE DEFINER='root'@'localhost' PROCEDURE mysql.add_flow_control ( IN sql_type INT, IN str VARCHAR(1024), IN max_num INT )    COMMENT 'Description-----------Basic functions for inserting rules.It is not recommended to call it directly, but to call it through add_select_flow_control、add_update_flow_control、add_update_flow_control and add_delete_flow_control.'    SQL SECURITY INVOKERBEGIN    IF (sql_type = 0) THEN        INSERT INTO mysql.du_flow_control_rules(type, max_concur, orig_str) VALUES('SELECT', max_num, str);    ELSEIF (sql_type = 1) THEN        INSERT INTO mysql.du_flow_control_rules(type, max_concur, orig_str) VALUES('UPDATE', max_num, str);    ELSEIF (sql_type = 2) THEN        INSERT INTO mysql.du_flow_control_rules(type, max_concur, orig_str) VALUES('INSERT', max_num, str);    ELSEIF (sql_type = 3) THEN         INSERT INTO mysql.du_flow_control_rules(type, max_concur, orig_str) VALUES('DELETE', max_num, str);    ELSE        SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'Sql type is error, please input correctly.';    END IF;END$$
CREATE DEFINER='root'@'localhost' PROCEDURE mysql.add_select_flow_control (IN str VARCHAR(1024), IN max_num INT )    COMMENT 'Description-----------Used to add select type rules to the current rule table.
Parameters-----------
str (VARCHAR(1024)):  The string of select rules entered by user.
max_num (INT):  The number of queries that can be executed concurrently.
Example--------
mysql> SELECT * FROM du_flow_control_rules;Empty set (0.00 sec)
mysql> CALL add_select_flow_control(''select~from~t1'', 100);Query OK, 1 row affected (0.00 sec)
mysql> SELECT * FROM du_flow_control_rules;+----+--------+------------+----------------+| id | type   | max_concur | orig_str       |+----+--------+------------+----------------+|  1 | SELECT |        100 | select~from~t1 |+----+--------+------------+----------------+1 row in set (0.00 sec)'    SQL SECURITY INVOKERBEGIN    CALL add_flow_control(0, str, max_num);END$$DELIMITER ;