1.1 MySQL简介
MySQL是世界上最流行的开源关系型数据库管理系统（RDBMS），由瑞典MySQL AB公司开发，现在属于Oracle公司。MySQL以其高性能、可靠性和易用性而闻名，广泛应用于Web应用程序、企业级应用和云计算环境中。
1.1.1 MySQL的特点
- 开源免费：MySQL社区版完全免费，源代码开放
- 跨平台：支持Linux、Windows、macOS等多种操作系统
- 高性能：优化的查询引擎，支持高并发访问
- 可扩展性：支持集群部署和水平扩展
- 安全性：提供完善的用户权限管理和数据加密
- 标准兼容：遵循SQL标准，易于学习和使用
1.1.2 MySQL版本历史
class MySQLVersionHistory:
def __init__(self):
self.versions = {
"3.23": "1999年 - 第一个稳定版本",
"4.0": "2003年 - 引入InnoDB存储引擎",
"4.1": "2004年 - 支持子查询和Unicode",
"5.0": "2005年 - 存储过程、触发器、视图",
"5.1": "2008年 - 分区表、插件架构",
"5.5": "2010年 - InnoDB成为默认引擎",
"5.6": "2013年 - 性能优化、在线DDL",
"5.7": "2015年 - JSON支持、性能提升",
"8.0": "2018年 - 窗口函数、CTE、角色管理"
}
def get_version_info(self, version):
return self.versions.get(version, "版本信息不存在")
def list_all_versions(self):
for version, info in self.versions.items():
print(f"MySQL {version}: {info}")
# 使用示例
version_history = MySQLVersionHistory()
version_history.list_all_versions()
print(f"\nMySQL 8.0特性: {version_history.get_version_info('8.0')}")
1.2 MySQL架构概述
MySQL采用分层架构设计，主要包括连接层、服务层、引擎层和存储层。
1.2.1 MySQL架构图
class MySQLArchitecture:
def __init__(self):
self.layers = {
"连接层": {
"组件": ["连接处理", "认证", "安全管理"],
"功能": "处理客户端连接，进行身份验证和权限检查"
},
"服务层": {
"组件": ["SQL解析器", "查询优化器", "缓存", "内置函数"],
"功能": "SQL语句解析、优化和执行计划生成"
},
"引擎层": {
"组件": ["InnoDB", "MyISAM", "Memory", "Archive"],
"功能": "数据存储和检索的具体实现"
},
"存储层": {
"组件": ["文件系统", "磁盘存储"],
"功能": "物理数据存储"
}
}
def describe_layer(self, layer_name):
if layer_name in self.layers:
layer = self.layers[layer_name]
print(f"\n{layer_name}:")
print(f"主要组件: {', '.join(layer['组件'])}")
print(f"核心功能: {layer['功能']}")
else:
print(f"未找到层级: {layer_name}")
def show_full_architecture(self):
print("MySQL完整架构:")
print("=" * 50)
for layer_name in self.layers.keys():
self.describe_layer(layer_name)
# 架构展示
architecture = MySQLArchitecture()
architecture.show_full_architecture()
1.2.2 连接层详解
连接层是MySQL架构的最上层，负责处理客户端连接和用户认证。
class MySQLConnectionLayer:
def __init__(self):
self.max_connections = 151 # 默认最大连接数
self.active_connections = 0
self.connection_pool = []
def establish_connection(self, user, password, host="localhost", port=3306):
"""建立数据库连接"""
if self.active_connections >= self.max_connections:
raise Exception("连接数已达上限")
connection_info = {
"connection_id": len(self.connection_pool) + 1,
"user": user,
"host": host,
"port": port,
"status": "active",
"created_time": "2024-01-15 10:00:00"
}
# 模拟认证过程
if self.authenticate_user(user, password):
self.connection_pool.append(connection_info)
self.active_connections += 1
print(f"连接建立成功: {connection_info['connection_id']}")
return connection_info
else:
raise Exception("认证失败")
def authenticate_user(self, user, password):
"""用户认证"""
# 简化的认证逻辑
valid_users = {
"root": "password123",
"admin": "admin123",
"user1": "user123"
}
return valid_users.get(user) == password
def close_connection(self, connection_id):
"""关闭连接"""
for conn in self.connection_pool:
if conn["connection_id"] == connection_id:
conn["status"] = "closed"
self.active_connections -= 1
print(f"连接 {connection_id} 已关闭")
break
def show_connection_status(self):
"""显示连接状态"""
print(f"\n当前活跃连接数: {self.active_connections}")
print(f"最大连接数: {self.max_connections}")
print("连接详情:")
for conn in self.connection_pool:
if conn["status"] == "active":
print(f" ID: {conn['connection_id']}, 用户: {conn['user']}, 主机: {conn['host']}")
# 连接管理示例
connection_layer = MySQLConnectionLayer()
# 建立连接
try:
conn1 = connection_layer.establish_connection("root", "password123")
conn2 = connection_layer.establish_connection("admin", "admin123")
connection_layer.show_connection_status()
except Exception as e:
print(f"连接失败: {e}")
1.2.3 服务层详解
服务层是MySQL的核心，包含SQL解析器、查询优化器等关键组件。
class MySQLServiceLayer:
def __init__(self):
self.query_cache = {}
self.parser = SQLParser()
self.optimizer = QueryOptimizer()
def process_sql(self, sql_statement):
"""处理SQL语句"""
print(f"\n处理SQL: {sql_statement}")
# 1. 检查查询缓存
if sql_statement in self.query_cache:
print("从缓存返回结果")
return self.query_cache[sql_statement]
# 2. SQL解析
parsed_sql = self.parser.parse(sql_statement)
print(f"解析结果: {parsed_sql}")
# 3. 查询优化
optimized_plan = self.optimizer.optimize(parsed_sql)
print(f"优化计划: {optimized_plan}")
# 4. 执行查询（模拟）
result = self.execute_query(optimized_plan)
# 5. 缓存结果
self.query_cache[sql_statement] = result
return result
def execute_query(self, execution_plan):
"""执行查询计划"""
# 模拟查询执行
return f"查询结果: {execution_plan['estimated_rows']} 行数据"
class SQLParser:
def parse(self, sql):
"""SQL解析器"""
sql_upper = sql.upper().strip()
if sql_upper.startswith('SELECT'):
return {
"type": "SELECT",
"tables": self.extract_tables(sql),
"columns": self.extract_columns(sql),
"where_clause": self.extract_where(sql)
}
elif sql_upper.startswith('INSERT'):
return {"type": "INSERT", "table": "users"}
elif sql_upper.startswith('UPDATE'):
return {"type": "UPDATE", "table": "users"}
elif sql_upper.startswith('DELETE'):
return {"type": "DELETE", "table": "users"}
else:
return {"type": "UNKNOWN"}
def extract_tables(self, sql):
# 简化的表名提取
return ["users", "orders"]
def extract_columns(self, sql):
# 简化的列名提取
return ["id", "name", "email"]
def extract_where(self, sql):
# 简化的WHERE子句提取
return "id > 100"
class QueryOptimizer:
def optimize(self, parsed_sql):
"""查询优化器"""
if parsed_sql["type"] == "SELECT":
return {
"execution_type": "index_scan",
"estimated_cost": 1.5,
"estimated_rows": 1000,
"indexes_used": ["idx_id", "idx_name"]
}
else:
return {
"execution_type": "table_scan",
"estimated_cost": 10.0,
"estimated_rows": 5000
}
# 服务层使用示例
service_layer = MySQLServiceLayer()
# 处理不同类型的SQL语句
sql_statements = [
"SELECT * FROM users WHERE id > 100",
"INSERT INTO users (name, email) VALUES ('张三', 'zhang@example.com')",
"UPDATE users SET email = 'new@example.com' WHERE id = 1",
"SELECT * FROM users WHERE id > 100" # 重复查询，测试缓存
]
for sql in sql_statements:
result = service_layer.process_sql(sql)
print(f"执行结果: {result}\n")
1.3 存储引擎
MySQL支持多种存储引擎，每种引擎都有其特定的用途和优势。
1.3.1 存储引擎对比
class MySQLStorageEngines:
def __init__(self):
self.engines = {
"InnoDB": {
"特性": ["事务支持", "外键约束", "行级锁", "崩溃恢复"],
"适用场景": "OLTP应用、高并发读写",
"优点": "ACID兼容、高并发性能好",
"缺点": "占用空间较大"
},
"MyISAM": {
"特性": ["表级锁", "全文索引", "压缩表"],
"适用场景": "读多写少、数据仓库",
"优点": "查询速度快、占用空间小",
"缺点": "不支持事务、不支持外键"
},
"Memory": {
"特性": ["内存存储", "哈希索引", "表级锁"],
"适用场景": "临时表、缓存",
"优点": "访问速度极快",
"缺点": "数据易丢失、容量受限"
},
"Archive": {
"特性": ["高压缩比", "只支持INSERT和SELECT"],
"适用场景": "日志存储、历史数据归档",
"优点": "压缩比高、存储成本低",
"缺点": "不支持索引、不支持UPDATE/DELETE"
}
}
def compare_engines(self):
"""对比存储引擎"""
print("MySQL存储引擎对比:")
print("=" * 80)
for engine_name, engine_info in self.engines.items():
print(f"\n{engine_name}:")
print(f" 特性: {', '.join(engine_info['特性'])}")
print(f" 适用场景: {engine_info['适用场景']}")
print(f" 优点: {engine_info['优点']}")
print(f" 缺点: {engine_info['缺点']}")
def recommend_engine(self, scenario):
"""根据场景推荐存储引擎"""
recommendations = {
"电商网站": "InnoDB - 支持事务，适合订单处理",
"数据分析": "MyISAM - 读取性能好，适合分析查询",
"会话存储": "Memory - 内存存储，访问速度快",
"日志归档": "Archive - 高压缩比，节省存储空间"
}
return recommendations.get(scenario, "请提供具体应用场景")
# 存储引擎使用示例
engines = MySQLStorageEngines()
engines.compare_engines()
print("\n场景推荐:")
scenarios = ["电商网站", "数据分析", "会话存储", "日志归档"]
for scenario in scenarios:
recommendation = engines.recommend_engine(scenario)
print(f"{scenario}: {recommendation}")
1.3.2 InnoDB存储引擎详解
InnoDB是MySQL的默认存储引擎，提供了完整的ACID事务支持。
class InnoDBEngine:
def __init__(self):
self.buffer_pool_size = "128MB" # 缓冲池大小
self.log_file_size = "48MB" # 日志文件大小
self.transaction_isolation = "REPEATABLE-READ" # 事务隔离级别
def transaction_demo(self):
"""事务处理示例"""
print("InnoDB事务处理示例:")
print("=" * 40)
# 模拟事务操作
transactions = [
{
"id": "TXN001",
"operations": [
"BEGIN",
"UPDATE accounts SET balance = balance - 100 WHERE id = 1",
"UPDATE accounts SET balance = balance + 100 WHERE id = 2",
"COMMIT"
],
"status": "SUCCESS"
},
{
"id": "TXN002",
"operations": [
"BEGIN",
"INSERT INTO orders (user_id, amount) VALUES (1, 500)",
"UPDATE inventory SET quantity = quantity - 1 WHERE product_id = 100",
"ROLLBACK" # 模拟回滚
],
"status": "ROLLBACK"
}
]
for txn in transactions:
print(f"\n事务 {txn['id']}:")
for op in txn['operations']:
print(f" {op}")
print(f" 状态: {txn['status']}")
def locking_mechanism(self):
"""锁机制演示"""
print("\nInnoDB锁机制:")
print("=" * 40)
lock_types = {
"行级锁": {
"共享锁(S)": "允许多个事务同时读取同一行",
"排他锁(X)": "只允许一个事务修改某一行"
},
"表级锁": {
"意向共享锁(IS)": "表示事务准备在表中加共享锁",
"意向排他锁(IX)": "表示事务准备在表中加排他锁"
},
"间隙锁": {
"Gap Lock": "锁定索引记录之间的间隙",
"Next-Key Lock": "行锁+间隙锁的组合"
}
}
for lock_category, locks in lock_types.items():
print(f"\n{lock_category}:")
for lock_name, description in locks.items():
print(f" {lock_name}: {description}")
def mvcc_explanation(self):
"""MVCC多版本并发控制"""
print("\nMVCC多版本并发控制:")
print("=" * 40)
mvcc_components = {
"隐藏列": {
"DB_TRX_ID": "创建该记录的事务ID",
"DB_ROLL_PTR": "回滚指针，指向undo log",
"DB_ROW_ID": "行ID（如果没有主键）"
},
"Read View": {
"m_ids": "当前活跃事务ID列表",
"min_trx_id": "最小活跃事务ID",
"max_trx_id": "下一个要分配的事务ID",
"creator_trx_id": "创建该Read View的事务ID"
},
"Undo Log": {
"作用": "存储数据的历史版本",
"类型": "insert undo log, update undo log"
}
}
for component, details in mvcc_components.items():
print(f"\n{component}:")
if isinstance(details, dict):
for key, value in details.items():
print(f" {key}: {value}")
else:
print(f" {details}")
# InnoDB引擎演示
innodb = InnoDBEngine()
innodb.transaction_demo()
innodb.locking_mechanism()
innodb.mvcc_explanation()
1.4 MySQL数据类型
MySQL支持多种数据类型，正确选择数据类型对性能和存储效率至关重要。
1.4.1 数值类型
class MySQLDataTypes:
def __init__(self):
self.numeric_types = {
"整数类型": {
"TINYINT": {"字节": 1, "范围": "(-128, 127)", "无符号范围": "(0, 255)"},
"SMALLINT": {"字节": 2, "范围": "(-32768, 32767)", "无符号范围": "(0, 65535)"},
"MEDIUMINT": {"字节": 3, "范围": "(-8388608, 8388607)", "无符号范围": "(0, 16777215)"},
"INT": {"字节": 4, "范围": "(-2147483648, 2147483647)", "无符号范围": "(0, 4294967295)"},
"BIGINT": {"字节": 8, "范围": "(-9223372036854775808, 9223372036854775807)", "无符号范围": "(0, 18446744073709551615)"}
},
"浮点类型": {
"FLOAT": {"字节": 4, "精度": "单精度", "范围": "±1.175494351E-38 到 ±3.402823466E+38"},
"DOUBLE": {"字节": 8, "精度": "双精度", "范围": "±2.2250738585072014E-308 到 ±1.7976931348623157E+308"}
},
"定点类型": {
"DECIMAL": {"特点": "精确数值", "用途": "货币计算", "语法": "DECIMAL(M,D)"}
}
}
def show_numeric_types(self):
"""显示数值类型"""
print("MySQL数值类型详解:")
print("=" * 60)
for category, types in self.numeric_types.items():
print(f"\n{category}:")
for type_name, properties in types.items():
print(f" {type_name}:")
for prop, value in properties.items():
print(f" {prop}: {value}")
def recommend_numeric_type(self, use_case):
"""根据用途推荐数值类型"""
recommendations = {
"用户ID": "INT UNSIGNED - 40亿用户足够使用",
"年龄": "TINYINT UNSIGNED - 0-255岁范围",
"价格": "DECIMAL(10,2) - 精确的货币计算",
"百分比": "DECIMAL(5,2) - 如99.99%",
"计数器": "BIGINT UNSIGNED - 大数值计数",
"评分": "DECIMAL(3,1) - 如9.5分"
}
return recommendations.get(use_case, "请提供具体使用场景")
# 数值类型演示
data_types = MySQLDataTypes()
data_types.show_numeric_types()
print("\n使用场景推荐:")
use_cases = ["用户ID", "年龄", "价格", "百分比", "计数器", "评分"]
for case in use_cases:
recommendation = data_types.recommend_numeric_type(case)
print(f"{case}: {recommendation}")
1.4.2 字符串类型
class MySQLStringTypes:
def __init__(self):
self.string_types = {
"定长字符串": {
"CHAR(M)": {
"长度": "0-255字符",
"存储": "固定长度，不足补空格",
"适用": "长度固定的数据，如手机号、身份证号"
}
},
"变长字符串": {
"VARCHAR(M)": {
"长度": "0-65535字符",
"存储": "变长存储，节省空间",
"适用": "长度不固定的数据，如姓名、地址"
}
},
"文本类型": {
"TINYTEXT": {"最大长度": "255字符", "用途": "短文本"},
"TEXT": {"最大长度": "65535字符", "用途": "中等长度文本"},
"MEDIUMTEXT": {"最大长度": "16777215字符", "用途": "较长文本"},
"LONGTEXT": {"最大长度": "4294967295字符", "用途": "超长文本"}
},
"二进制类型": {
"BINARY(M)": {"特点": "定长二进制", "用途": "存储二进制数据"},
"VARBINARY(M)": {"特点": "变长二进制", "用途": "存储变长二进制数据"},
"BLOB": {"特点": "二进制大对象", "用途": "存储图片、文件等"}
}
}
def show_string_types(self):
"""显示字符串类型"""
print("MySQL字符串类型详解:")
print("=" * 60)
for category, types in self.string_types.items():
print(f"\n{category}:")
for type_name, properties in types.items():
print(f" {type_name}:")
for prop, value in properties.items():
print(f" {prop}: {value}")
def char_vs_varchar_demo(self):
"""CHAR vs VARCHAR 对比演示"""
print("\nCHAR vs VARCHAR 存储对比:")
print("=" * 50)
test_data = [
{"value": "ABC", "char_10": "'ABC '", "varchar_10": "'ABC'"},
{"value": "HELLO", "char_10": "'HELLO '", "varchar_10": "'HELLO'"},
{"value": "1234567890", "char_10": "'1234567890'", "varchar_10": "'1234567890'"}
]
print(f"{'原始值':<12} {'CHAR(10)':<15} {'VARCHAR(10)':<15} {'存储字节':<10}")
print("-" * 60)
for data in test_data:
char_bytes = 10 # CHAR固定长度
varchar_bytes = len(data["value"]) + 1 # VARCHAR长度+1字节长度信息
print(f"{data['value']:<12} {data['char_10']:<15} {data['varchar_10']:<15} CHAR:{char_bytes}, VARCHAR:{varchar_bytes}")
def text_type_selection(self):
"""文本类型选择指南"""
print("\n文本类型选择指南:")
print("=" * 40)
scenarios = {
"用户评论": "TEXT - 一般评论长度适中",
"商品描述": "TEXT 或 MEDIUMTEXT - 根据描述长度",
"文章内容": "MEDIUMTEXT 或 LONGTEXT - 文章较长",
"日志记录": "TEXT - 日志信息中等长度",
"配置信息": "VARCHAR(1000) - 配置信息相对固定"
}
for scenario, recommendation in scenarios.items():
print(f"{scenario}: {recommendation}")
# 字符串类型演示
string_types = MySQLStringTypes()
string_types.show_string_types()
string_types.char_vs_varchar_demo()
string_types.text_type_selection()
1.4.3 日期时间类型
from datetime import datetime, date, time
class MySQLDateTimeTypes:
def __init__(self):
self.datetime_types = {
"DATE": {
"格式": "YYYY-MM-DD",
"范围": "1000-01-01 到 9999-12-31",
"存储字节": 3,
"用途": "存储日期"
},
"TIME": {
"格式": "HH:MM:SS",
"范围": "-838:59:59 到 838:59:59",
"存储字节": 3,
"用途": "存储时间"
},
"DATETIME": {
"格式": "YYYY-MM-DD HH:MM:SS",
"范围": "1000-01-01 00:00:00 到 9999-12-31 23:59:59",
"存储字节": 8,
"用途": "存储日期和时间"
},
"TIMESTAMP": {
"格式": "YYYY-MM-DD HH:MM:SS",
"范围": "1970-01-01 00:00:01 到 2038-01-19 03:14:07",
"存储字节": 4,
"用途": "存储时间戳，自动更新"
},
"YEAR": {
"格式": "YYYY",
"范围": "1901 到 2155",
"存储字节": 1,
"用途": "存储年份"
}
}
def show_datetime_types(self):
"""显示日期时间类型"""
print("MySQL日期时间类型详解:")
print("=" * 60)
for type_name, properties in self.datetime_types.items():
print(f"\n{type_name}:")
for prop, value in properties.items():
print(f" {prop}: {value}")
def datetime_vs_timestamp(self):
"""DATETIME vs TIMESTAMP 对比"""
print("\nDATETIME vs TIMESTAMP 对比:")
print("=" * 50)
comparison = {
"存储范围": {
"DATETIME": "1000年到9999年",
"TIMESTAMP": "1970年到2038年"
},
"时区处理": {
"DATETIME": "不处理时区，存储什么显示什么",
"TIMESTAMP": "根据时区自动转换"
},
"自动更新": {
"DATETIME": "不自动更新",
"TIMESTAMP": "可设置自动更新"
},
"存储空间": {
"DATETIME": "8字节",
"TIMESTAMP": "4字节"
}
}
for aspect, details in comparison.items():
print(f"\n{aspect}:")
for type_name, description in details.items():
print(f" {type_name}: {description}")
def datetime_functions_demo(self):
"""日期时间函数演示"""
print("\n常用日期时间函数:")
print("=" * 40)
functions = {
"NOW()": "返回当前日期和时间",
"CURDATE()": "返回当前日期",
"CURTIME()": "返回当前时间",
"DATE(datetime)": "提取日期部分",
"TIME(datetime)": "提取时间部分",
"YEAR(date)": "提取年份",
"MONTH(date)": "提取月份",
"DAY(date)": "提取日期",
"HOUR(time)": "提取小时",
"MINUTE(time)": "提取分钟",
"SECOND(time)": "提取秒数",
"DATE_ADD(date, INTERVAL expr unit)": "日期加法",
"DATE_SUB(date, INTERVAL expr unit)": "日期减法",
"DATEDIFF(date1, date2)": "计算日期差",
"DATE_FORMAT(date, format)": "格式化日期"
}
for func, description in functions.items():
print(f"{func}: {description}")
# 示例SQL
print("\n示例SQL:")
examples = [
"SELECT NOW();",
"SELECT DATE_ADD('2024-01-15', INTERVAL 30 DAY);",
"SELECT DATEDIFF('2024-12-31', '2024-01-01');",
"SELECT DATE_FORMAT(NOW(), '%Y年%m月%d日');"
]
for example in examples:
print(f" {example}")
# 日期时间类型演示
datetime_types = MySQLDateTimeTypes()
datetime_types.show_datetime_types()
datetime_types.datetime_vs_timestamp()
datetime_types.datetime_functions_demo()
1.5 总结
本章介绍了MySQL的基础概念和架构，包括：
核心要点
- MySQL特点：开源免费、跨平台、高性能、可扩展
- 分层架构：连接层、服务层、引擎层、存储层
- 存储引擎：InnoDB（默认）、MyISAM、Memory、Archive
- 数据类型：数值类型、字符串类型、日期时间类型
最佳实践
存储引擎选择：
- OLTP应用优选InnoDB
- 读多写少场景可考虑MyISAM
- 临时数据使用Memory
- 归档数据使用Archive
数据类型选择：
- 根据数据范围选择合适的整数类型
- 货币计算使用DECIMAL
- 变长字符串优选VARCHAR
- 根据时区需求选择DATETIME或TIMESTAMP
性能优化：
- 合理设置连接数
- 启用查询缓存
- 选择合适的存储引擎
- 优化数据类型选择
下一章我们将学习MySQL的安装部署与环境配置，包括不同操作系统的安装方法、配置优化和安全设置。
