在之前的MySQL运维中，要求禁用触发器/存储过程/外键等一些数据库常见功能，因此对MySQL外键也相对比较陌生，今天特地探究下。
现有表TB001和TB002各包含6291456行数据，创建脚本如下：
CREATE TABLE `tb001` ( `ID` int(11) NOT NULL AUTO_INCREMENT, `C1` int(11) DEFAULT NULL, PRIMARY KEY (`ID`) ) ENGINE=InnoDB AUTO_INCREMENT=6684558 DEFAULT CHARSET=utf8; CREATE TABLE `tb002` ( `ID` int(11) NOT NULL AUTO_INCREMENT, `C1` int(11) DEFAULT NULL, `C2` int(11) DEFAULT NULL, PRIMARY KEY (`ID`) ) ENGINE=InnoDB AUTO_INCREMENT=6356897 DEFAULT CHARSET=utf8;
现在在TB002上的C1列增加一个外键指向TB001的主键，有如下三种创建方式：
方式1：直接创建外键
查询：ALTER TABLE TB002 ADD CONSTRAINT `FK_C1` FOREIGN KEY (`C1`) REFERENCES `TB001` (`ID`) 共 6291456 行受到影响 执行耗时 : 55.889 sec 传送时间 : 1.003 sec 总耗时 : 56.892 sec
方式2：创建索引+创建外键
查询：alter table TB002 ADD INDEX FK_C1(C1) 共 0 行受到影响 执行耗时 : 9.676 sec 传送时间 : 1.001 sec 总耗时 : 10.678 sec -------------------------------------------------- 查询：ALTER TABLE TB002 ADD CONSTRAINT `FK_C1` FOREIGN KEY (`C1`) REFERENCES `TB001` (`ID`) 共 6291458 行受到影响 执行耗时 : 1 min 9 sec 传送时间 : 1.001 sec 总耗时 : 1 min 10 sec
方式3：创建索引+创建外键+FOREIGN_KEY_CHECKS
查询：alter table TB002 ADD INDEX FK_C1(C1) 共 0 行受到影响 执行耗时 : 9.393 sec 传送时间 : 1.002 sec 总耗时 : 10.396 sec -------------------------------------------------- 查询：SET @@foreign_key_checks=0 共 0 行受到影响 执行耗时 : 0 sec 传送时间 : 0 sec 总耗时 : 0 sec -------------------------------------------------- 查询：ALTER TABLE TB002 ADD CONSTRAINT `FK_C1` FOREIGN KEY (`C1`) REFERENCES `TB001` (`ID`) 共 0 行受到影响 执行耗时 : 0.002 sec 传送时间 : 0.001 sec 总耗时 : 0.004 sec -------------------------------------------------- 查询：SET @@foreign_key_checks=1 共 0 行受到影响 执行耗时 : 0 sec 传送时间 : 0 sec 总耗时 : 0 sec
如未设置FOREIGN_KEY_CHECKS=0，在创建外键过程中，会阻塞其他会话对表TB002的增删改操作(删除操作也不允许)。
从上面三种方式执行结果可以发现，创建外键最快的方式是：创建索引+创建外键+FOREIGN_KEY_CHECKS，在设置FOREIGN_KEY_CHECKS=0时，创建外键不会检查当前表中数据，属于元数据操作，因此很快完成，对其他会话影响较小，但无法保证表中数据满足外键约束。
当使用pt-online-schema-change修改被外键依赖的表时，如果参数alter-foreign-keys-method指定为drop_swap，那么在日志中可以看到rename被依赖表名前，会使用FOREIGN_KEY_CHECKS来屏蔽检查：
2019-10-10T07:48:06.692791-05:00 3428 Query SHOW WARNINGS 2019-10-10T07:48:06.693796-05:00 3428 Query SHOW GLOBAL STATUS LIKE 'Threads_running' 2019-10-10T07:48:06.695144-05:00 3428 Query ANALYZE TABLE `dadaabc_test`.`_tb001_new` /* pt-online-schema-change */ 2019-10-10T07:48:06.706498-05:00 3428 Query SET foreign_key_checks=0 2019-10-10T07:48:06.707180-05:00 3428 Query DROP TABLE IF EXISTS `dadaabc_test`.`tb001` 2019-10-10T07:48:06.731187-05:00 3428 Query RENAME TABLE `dadaabc_test`.`_tb001_new` TO `dadaabc_test`.`tb001` 2019-10-10T07:48:06.734388-05:00 3428 Query DROP TRIGGER IF EXISTS `dadaabc_test`.`pt_osc_dadaabc_test_tb001_del` 2019-10-10T07:48:06.736416-05:00 3428 Query DROP TRIGGER IF EXISTS `dadaabc_test`.`pt_osc_dadaabc_test_tb001_upd` 2019-10-10T07:48:06.736893-05:00 3428 Query DROP TRIGGER IF EXISTS `dadaabc_test`.`pt_osc_dadaabc_test_tb001_ins` 2019-10-10T07:48:06.737513-05:00 3428 Query SHOW TABLES FROM `dadaabc_test` LIKE '\_tb001\_new'
drop_swap会删除掉原表，再将新表改名，如果在删除原表之后重命名新表完成之前出现故障，会导致应用长时间异常。
FOREIGN_KEY_CHECKS相关
FOREIGN_KEY_CHECKS在官方文档解释：
If set to 1 (the default), foreign key constraints for InnoDB tables are checked. If set to 0, foreign key constraints are ignored, with a couple of exceptions. When re-creating a table that was dropped, an error is returned if the table definition does not conform to the foreign key constraints referencing the table. Likewise, an ALTER TABLE operation returns an error if a foreign key definition is incorrectly formed. Setting foreign_key_checks to 0 also affects data definition statements: DROP SCHEMA drops a schema even if it contains tables that have foreign keys that are referred to by tables outside the schema, and DROP TABLE drops tables that have foreign keys that are referred to by other tables. Setting foreign_key_checks to 1 does not trigger a scan of the existing table data. Therefore, rows added to the table while foreign_key_checks=0 will not be verified for consistency.
当FOREIGN_KEY_CHECKS设置为0时，外键约束会被忽略，而当FOREIGN_KEY_CHECKS设置为1时不会触发约束检查。
删除外键
## 删除外键，但外键依赖的索引不会被删除 ALTER TABLE TB002 DROP FOREIGN KEY FK_C1; ## 删除外键依赖的索引 ALTER TABLE TB002 DROP INDEX FK_C1;
导出实例下的外键
# coding: utf-8 import pymysql import os, traceback, datetime pymysql.install_as_MySQLdb() import MySQLdb working_folder = os.path.dirname(os.path.abspath(__file__)) default_log_file = os.path.join(working_folder, "default_log.txt") error_log_file = os.path.join(working_folder, "error_log.txt") create_key_file = os.path.join(working_folder, "create_key_script.txt") drop_key_file = os.path.join(working_folder, "drop_key_script.txt") class MySQLForeignKeyLoader(object): def __init__(self, mysql_host, mysql_port, mysql_user, mysql_password, mysql_charset, database_name, connect_timeout=60): self.mysql_host = mysql_host self.mysql_port = mysql_port self.mysql_user = mysql_user self.mysql_password = mysql_password self.mysql_charset = mysql_charset self.database_name = database_name self.connect_timeout = connect_timeout def highlight(self, message): return "%s[30;2m%s%s[1m" % (chr(27), message, chr(27)) def print_warning_message(self, message): """ 以红色字体显示消息内容 :param message: 消息内容 :return: 无返回值 """ message = str(message) print(self.highlight('') + "%s[31;1m%s%s[0m" % (chr(27), message, chr(27))) self.write_file(error_log_file, message) def print_info_message(self, message): """ 以绿色字体输出提醒性的消息 :param message: 消息内容 :return: 无返回值 """ message = str(message) print(self.highlight('') + "%s[32;2m%s%s[0m" % (chr(27), message, chr(27))) self.write_file(default_log_file, message) def write_file(self, file_path, message): """ 将传入的message追加写入到file_path指定的文件中 请先创建文件所在的目录 :param file_path: 要写入的文件路径 :param message: 要写入的信息 :return: """ file_handle = open(file_path, 'a') file_handle.writelines(message) # 追加一个换行以方便浏览 file_handle.writelines(chr(10)) file_handle.flush() file_handle.close() def get_mysql_connection(self, is_use_dict=False): """ 根据默认配置返回数据库连接 :return: 数据库连接 """ if is_use_dict: conn = MySQLdb.connect( host=self.mysql_host, port=self.mysql_port, user=self.mysql_user, passwd=self.mysql_password, connect_timeout=self.connect_timeout, charset=self.mysql_charset, db=self.database_name, cursorclass=pymysql.cursors.DictCursor ) else: conn = MySQLdb.connect( host=self.mysql_host, port=self.mysql_port, user=self.mysql_user, passwd=self.mysql_password, connect_timeout=self.connect_timeout, charset=self.mysql_charset, db=self.database_name, cursorclass=pymysql.cursors.Cursor ) return conn def mysql_query(self, sql_script, sql_paras=None, is_use_dict=True): """ 执行SQL :param sql_script: :param sql_paras: :param is_use_dict: :return: """ cursor = None mysql_conn = None try: mysql_conn = self.get_mysql_connection(is_use_dict=is_use_dict) cursor = mysql_conn.cursor() if sql_paras is not None: cursor.execute(sql_script, sql_paras) else: cursor.execute(sql_script) exec_result = cursor.fetchall() return exec_result except Exception as ex: warning_message = """ mysql query exception, sql script:{sql_script} sql paras:{sql_paras} exception:{sql_exception} trace back:{sql_trace} """.format( sql_script=sql_script, sql_paras=str(sql_paras), sql_exception=str(ex), sql_trace=str(traceback.format_exc()) ) self.print_warning_message(warning_message) raise Exception(str(ex)) finally: if cursor is not None: cursor.close() if mysql_conn is not None: mysql_conn.close() def mysql_exec(self, sql_script, sql_paras=None): """ 执行SQL :param sql_script: :param sql_paras: :return: """ cursor = None mysql_conn = None try: mysql_conn = self.get_mysql_connection(is_use_dict=False) cursor = mysql_conn.cursor() if sql_paras is not None: cursor.execute(sql_script, sql_paras) else: cursor.execute(sql_script) affect_rows = cursor.rowcount mysql_conn.commit() return affect_rows except Exception as ex: warning_message = """ mysql query exception, sql script:{sql_script} sql paras:{sql_paras} exception:{sql_exception} trace back:{sql_trace} """.format( sql_script=sql_script, sql_paras=str(sql_paras), sql_exception=str(ex), sql_trace=str(traceback.format_exc()) ) self.print_warning_message(warning_message) raise Exception(str(ex)) finally: if cursor is not None: cursor.close() if mysql_conn is not None: mysql_conn.close() def get_foreign_keys(self): sql_script = """ SELECT * FROM INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS WHERE CONSTRAINT_SCHEMA NOT IN('sys','mysql','information_schema','proformance_schema') """ return self.mysql_query(sql_script=sql_script, is_use_dict=True) def get_drop_foreign_key_script(self, source_database_name, source_table_name, foreign_key_name): sql_script = "ALTER TABLE `{database_name}`.`{table_name}` DROP FOREIGN KEY `{foreign_key_name}`;".format( database_name=source_database_name, table_name=source_table_name, foreign_key_name=foreign_key_name ) self.print_info_message(sql_script) return sql_script def get_create_table_script(self, source_database_name, source_table_name): sql_script = "SHOW CREATE TABLE `{database_name}`.`{table_name}`;".format( database_name=source_database_name, table_name=source_table_name ) query_result = self.mysql_query(sql_script=sql_script, is_use_dict=False) if len(query_result) > 0: return query_result[0][1] else: return "" def get_create_foreign_key_script(self, source_database_name, source_table_name, foreign_key_name): table_script = self.get_create_table_script( source_database_name=source_database_name, source_table_name=source_table_name ) table_script = table_script.encode("utf-8") script_items = table_script.splitlines() constraint_script = "" for script_item in script_items: if script_item.lower().find(str(foreign_key_name).lower()) > 0 and script_item.lower().find(" FOREIGN KEY ".lower()) > 0 and script_item.lower().find(" CONSTRAINT ".lower()) > 0 and script_item.lower().find(" REFERENCES ".lower()) > 0: constraint_script = script_item break if constraint_script <> "": foreign_key_script = "ALTER TABLE `{database_name}`.`{table_name}` ADD {constraint_script};".format( database_name=source_database_name, table_name=source_table_name, constraint_script=constraint_script ) else: foreign_key_script = "## Can not load script for foreign key `{foreign_key_name}` in `{database_name}`.`{table_name}`;".format( database_name=source_database_name, table_name=source_table_name, foreign_key_name=foreign_key_name ) self.print_info_message(foreign_key_script) return foreign_key_script def load_foreign_key_scripts(self): self.print_info_message("start to load script for foreign keys") foreign_key_list = self.get_foreign_keys() for foreign_key_item in foreign_key_list: source_database_name = foreign_key_item["CONSTRAINT_SCHEMA"] source_table_name = foreign_key_item["TABLE_NAME"] foreign_key_name = foreign_key_item["CONSTRAINT_NAME"] self.print_info_message( "load script for foreign key `{foreign_key_name}` in `{source_database_name}`.`{source_table_name}`".format( source_database_name=source_database_name, source_table_name=source_table_name, foreign_key_name=foreign_key_name )) drop_script = self.get_drop_foreign_key_script( source_database_name=source_database_name, source_table_name=source_table_name, foreign_key_name=foreign_key_name ) self.write_file(drop_key_file, drop_script) create_script = self.get_create_foreign_key_script( source_database_name=source_database_name, source_table_name=source_table_name, foreign_key_name=foreign_key_name ) self.write_file(create_key_file, create_script) self.print_info_message("end to load script for foreign keys") def main(): loader = MySQLForeignKeyLoader( mysql_host="192.168.0.1", mysql_port=3306, mysql_user="root", mysql_password="root", mysql_charset="utf8", database_name="mysql" ) loader.load_foreign_key_scripts() if __name__ == '__main__': main()