# 实体类与数据库脚本同步参考手册 ## 详细映射规则 ### 数据类型映射表 #### 基本数据类型 | Java类型 | MySQL类型 | 注解示例 | 说明 | |----------|-----------|----------|------| | `String` | `VARCHAR(N)` | `@Column(length=50)` | 默认长度255 | | `String` | `TEXT` | `@Column(length=2000)` | 长文本 | | `String` | `LONGTEXT` | `@Column(length=10000)` | 超长文本 | | `Integer`/`int` | `INT` | `@Column` | 32位整数 | | `Long`/`long` | `BIGINT` | `@Column` | 64位整数 | | `Short`/`short` | `SMALLINT` | `@Column` | 16位整数 | | `Byte`/`byte` | `TINYINT` | `@Column` | 8位整数 | | `Float`/`float` | `FLOAT` | `@Column` | 单精度浮点 | | `Double`/`double` | `DOUBLE` | `@Column` | 双精度浮点 | | `Boolean`/`boolean` | `TINYINT(1)` | `@Column` | 布尔值 | | `Character`/`char` | `CHAR(1)` | `@Column` | 单字符 | #### 数值类型 | Java类型 | MySQL类型 | 注解示例 | 说明 | |----------|-----------|----------|------| | `BigDecimal` | `DECIMAL(M,D)` | `@Column(precision=10, scale=2)` | 精确小数 | | `BigInteger` | `DECIMAL(M,0)` | `@Column(precision=20)` | 大整数 | #### 日期时间类型 | Java类型 | MySQL类型 | 注解示例 | 说明 | |----------|-----------|----------|------| | `LocalDate` | `DATE` | `@Column` | 日期 | | `LocalTime` | `TIME` | `@Column` | 时间 | | `LocalDateTime` | `DATETIME` | `@Column` | 日期时间 | | `Instant` | `TIMESTAMP` | `@Column` | 时间戳 | | `Year` | `YEAR` | `@Column` | 年份 | #### 枚举类型 ```java // 方式1: 字符串存储 @Enumerated(EnumType.STRING) @Column(length = 20) private UserType userType; // 方式2: 整数存储 @Enumerated(EnumType.ORDINAL) @Column private Status status; ``` #### 集合类型 ```java // JSON格式存储 @JdbcTypeCode(SqlTypes.JSON) @Column(columnDefinition = "JSON") private List tags; // 或者使用单独的关联表 @ElementCollection @CollectionTable(name = "user_roles", joinColumns = @JoinColumn(name = "user_id")) private Set roles; ``` ### 注解详细说明 #### @Column注解参数 ```java @Column( name = "column_name", // 数据库列名 length = 255, // VARCHAR长度 nullable = true, // 是否允许NULL unique = false, // 是否唯一 precision = 10, // 数字总位数 scale = 2, // 小数位数 columnDefinition = "TEXT" // 自定义列定义 ) ``` #### @Id相关注解 ```java // 自增主键 @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private Long id; // UUID主键 @Id @GeneratedValue(generator = "uuid2") @GenericGenerator(name = "uuid2", strategy = "uuid2") @Column(name = "id", length = 36) private String id; // 复合主键 @EmbeddedId private UserRoleId id; ``` #### 时间戳注解 ```java // 创建时间自动设置 @CreationTimestamp @Column(name = "created_time", updatable = false) private LocalDateTime createdTime; // 更新时间自动设置 @UpdateTimestamp @Column(name = "updated_time") private LocalDateTime updatedTime; ``` ## 变更管理工具 ### 变更脚本生成器 ```python #!/usr/bin/env python3 # generate-change-scripts.py import os import re import datetime from typing import Dict, List class ChangeScriptGenerator: def __init__(self, project_root: str): self.project_root = project_root # 部署目录结构：与开发代码分离 self.deploy_dir = os.path.join(project_root, "deploy", "sql") self.upgrade_dir = os.path.join(self.deploy_dir, "upgrade") self.init_dir = os.path.join(self.deploy_dir, "init") def create_version_directory(self, version: str) -> str: """创建版本目录""" version_dir = os.path.join(self.upgrade_dir, version) os.makedirs(version_dir, exist_ok=True) os.makedirs(os.path.join(version_dir, "rollback"), exist_ok=True) return version_dir def generate_alter_script(self, table_name: str, changes: Dict, version: str) -> str: """生成ALTER TABLE脚本""" today = datetime.date.today().strftime('%Y%m%d') timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 生成正向变更脚本 forward_content = f"""-- ============================================ -- 变更描述: {changes.get('description', '数据库结构变更')} -- 变更时间: {timestamp} -- 影响版本: {version} -- 影响表: {table_name} -- ============================================ """ # 添加字段变更 if 'add_columns' in changes: forward_content += "-- 1. 添加字段\n" for column_def in changes['add_columns']: forward_content += f"ALTER TABLE `{table_name}` ADD COLUMN {column_def};\n" forward_content += "\n" if 'modify_columns' in changes: forward_content += "-- 2. 修改字段\n" for column_def in changes['modify_columns']: forward_content += f"ALTER TABLE `{table_name}` MODIFY COLUMN {column_def};\n" forward_content += "\n" if 'drop_columns' in changes: forward_content += "-- 3. 删除字段\n" for column_name in changes['drop_columns']: forward_content += f"ALTER TABLE `{table_name}` DROP COLUMN `{column_name}`;\n" forward_content += "\n" # 添加索引变更 if 'add_indexes' in changes: forward_content += "-- 4. 添加索引\n" for index_def in changes['add_indexes']: forward_content += f"ALTER TABLE `{table_name}` ADD {index_def};\n" forward_content += "\n" # 生成回滚脚本 rollback_content = self.generate_rollback_script(table_name, changes, timestamp, version) # 保存脚本文件 version_dir = self.create_version_directory(version) forward_filename = f"{today}_{self.sanitize_filename(changes.get('description', 'change'))}.sql" rollback_filename = f"{today}_{self.sanitize_filename(changes.get('description', 'change'))}_rollback.sql" forward_path = os.path.join(version_dir, forward_filename) rollback_path = os.path.join(version_dir, "rollback", rollback_filename) with open(forward_path, 'w', encoding='utf-8') as f: f.write(forward_content) with open(rollback_path, 'w', encoding='utf-8') as f: f.write(rollback_content) return forward_path def generate_rollback_script(self, table_name: str, changes: Dict, timestamp: str, version: str) -> str: """生成回滚脚本""" rollback_content = f"""-- ============================================ -- 回滚描述: 撤销 {changes.get('description', '数据库变更')} -- 回滚时间: {timestamp} -- 原版本: {version} -- 影响表: {table_name} -- ============================================ """ # 删除索引（逆序） if 'add_indexes' in changes: rollback_content += "-- 1. 删除新增的索引\n" for index_def in reversed(changes['add_indexes']): index_name = self.extract_index_name(index_def) if index_name: rollback_content += f"ALTER TABLE `{table_name}` DROP INDEX `{index_name}`;\n" rollback_content += "\n" # 删除字段（逆序） if 'add_columns' in changes: rollback_content += "-- 2. 删除新增的字段\n" for column_def in reversed(changes['add_columns']): column_name = self.extract_column_name(column_def) if column_name: rollback_content += f"ALTER TABLE `{table_name}` DROP COLUMN `{column_name}`;\n" rollback_content += "\n" # 恢复修改的字段 if 'modify_columns' in changes: rollback_content += "-- 3. 恢复修改的字段\n" # 这里需要保存原始定义才能准确回滚 rollback_content += "-- TODO: 需要手动添加原始字段定义\n" rollback_content += "\n" return rollback_content def sanitize_filename(self, text: str) -> str: """清理文件名""" return re.sub(r'[^\w\-_]', '_', text.lower()) def extract_column_name(self, column_def: str) -> str: """从列定义中提取列名""" match = re.search(r'`([^`]+)`', column_def) return match.group(1) if match else "" def extract_index_name(self, index_def: str) -> str: """从索引定义中提取索引名""" match = re.search(r'`([^`]+)`\s*\(', index_def) return match.group(1) if match else "" # 使用示例 if __name__ == "__main__": generator = ChangeScriptGenerator("/path/to/project") changes = { 'description': '为用户表添加手机号字段', 'add_columns': [ '`phone` VARCHAR(20) COMMENT \'手机号码\'', '`wechat_id` VARCHAR(50) COMMENT \'微信ID\'' ], 'add_indexes': [ 'INDEX `idx_phone` (`phone`)', 'INDEX `idx_wechat_id` (`wechat_id`)' ] } script_path = generator.generate_alter_script('user', changes, 'v1.1.0') print(f"生成变更脚本: {script_path}") ``` ### 文档同步工具 ```python #!/usr/bin/env python3 # sync-documentation.py import os import json from typing import Dict, List class DocumentationSync: def __init__(self, project_root: str): self.project_root = project_root self.docs_dir = os.path.join(project_root, "docs") self.db_docs_dir = os.path.join(self.docs_dir, "database") def update_table_documentation(self, table_name: str, entity_info: Dict, changes: Dict): """更新表结构文档""" doc_file = os.path.join(self.db_docs_dir, f"{table_name}.md") # 读取现有文档或创建新文档 if os.path.exists(doc_file): with open(doc_file, 'r', encoding='utf-8') as f: content = f.read() else: content = self.create_table_template(table_name) # 更新字段信息 content = self.update_fields_section(content, entity_info['fields']) # 添加变更历史 content = self.add_change_history(content, changes) # 保存更新后的文档 os.makedirs(os.path.dirname(doc_file), exist_ok=True) with open(doc_file, 'w', encoding='utf-8') as f: f.write(content) def create_table_template(self, table_name: str) -> str: """创建表文档模板""" return f"""# {table_name} 表结构文档 ## 表基本信息 - **表名**: `{table_name}` - **描述**: - **引擎**: InnoDB - **字符集**: utf8mb4 ## 字段定义 | 字段名 | 类型 | 允许NULL | 默认值 | 描述 | |--------|------|----------|--------|------| | id | BIGINT | NO | AUTO_INCREMENT | 主键ID | ## 索引信息 | 索引名 | 类型 | 字段 | 描述 | |--------|------|------|------| | PRIMARY | PRIMARY | id | 主键索引 | ## 变更历史 | 日期 | 版本 | 变更描述 | 变更人 | |------|------|----------|--------| """ def update_fields_section(self, content: str, fields: List[Dict]) -> str: """更新字段定义部分""" # 找到字段定义表格的位置并替换 lines = content.split('\n') in_fields_section = False field_lines = [] for line in lines: if line.strip() == '## 字段定义': in_fields_section = True field_lines.append(line) continue if in_fields_section and line.startswith('|') and '字段名' in line: # 保留表头 field_lines.append(line) continue if in_fields_section and line.startswith('|----'): # 保留分隔线 field_lines.append(line) continue if in_fields_section and line.startswith('|') and line.count('|') >= 5: # 跳过旧的字段行 continue if in_fields_section and not line.strip(): # 结束字段部分 in_fields_section = False # 添加新的字段定义 for field in fields: field_lines.append(self.format_field_row(field)) field_lines.append('') field_lines.append(line) return '\n'.join(field_lines) def format_field_row(self, field: Dict) -> str: """格式化字段行""" nullable = 'YES' if field.get('nullable', True) else 'NO' default_val = field.get('default', '') description = field.get('comment', '') return f"| {field['name']} | {field['type']} | {nullable} | {default_val} | {description} |" def add_change_history(self, content: str, changes: Dict) -> str: """添加变更历史""" import datetime today = datetime.date.today().strftime('%Y-%m-%d') change_line = f"| {today} | {changes.get('version', 'unknown')} | {changes.get('description', '')} | {changes.get('author', 'system')} |" # 在变更历史部分添加新行 lines = content.split('\n') for i, line in enumerate(lines): if '| 日期 | 版本 | 变更描述 | 变更人 |' in line: # 在表头后插入新行 lines.insert(i + 2, change_line) break return '\n'.join(lines) # 配置文件示例 CONFIG = { "documentation": { "enabled": True, "output_dir": "docs/database", "auto_update": True }, "change_scripts": { "upgrade_dir": "src/main/resources/sql/upgrade", "version_pattern": "v{major}.{minor}.{patch}", "naming_convention": "{date}_{description}.sql" } } ``` ## 生产环境部署规范 ### 变更脚本目录结构 ``` project/ ├── deploy/ # 部署相关文件（与开发代码分离） │ └── sql/ │ ├── init/ # 初始部署脚本 │ │ ├── v1.0.0/ │ │ │ ├── 01_create_tables.sql │ │ │ ├── 02_insert_init_data.sql │ │ │ └── README.md │ │ └── v1.1.0/ │ │ └── 01_upgrade_v1.1.0.sql │ └── upgrade/ # 增量变更脚本 │ ├── v1.1.0/ │ │ ├── 20240115_add_user_phone.sql │ │ ├── 20240115_add_user_phone_rollback.sql │ │ └── changelog.md │ ├── v1.2.0/ │ │ ├── 20240201_modify_order_amount.sql │ │ ├── 20240201_modify_order_amount_rollback.sql │ │ └── changelog.md │ └── current/ # 当前开发中的变更 │ └── draft_changes/ └── src/main/resources/ # 开发代码目录（保持独立） ``` # 生产环境部署规范 ### 变更脚本执行流程 ```bash #!/bin/bash # deploy-database-changes.sh ENVIRONMENT=${1:-"test"} VERSION=${2:-"latest"} BACKUP_FIRST=${3:-"true"} SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" PROJECT_ROOT="${SCRIPT_DIR}/../.." # 使用部署目录 DEPLOY_SQL_DIR="${PROJECT_ROOT}/deploy/sql" echo "=== 数据库变更部署 ===" echo "环境: ${ENVIRONMENT}" echo "版本: ${VERSION}" echo "备份: ${BACKUP_FIRST}" echo "" # 1. 数据库备份 if [ "${BACKUP_FIRST}" = "true" ]; then echo "1. 执行数据库备份..." BACKUP_FILE="/tmp/db_backup_$(date +%Y%m%d_%H%M%S).sql" mysqldump -h${DB_HOST} -u${DB_USER} -p${DB_PASSWORD} ${DB_NAME} > "${BACKUP_FILE}" echo "备份完成: ${BACKUP_FILE}" echo "" fi # 2. 确定要执行的脚本 if [ "${VERSION}" = "latest" ]; then # 执行所有未应用的变更 CHANGE_SCRIPTS=$(find "${DEPLOY_SQL_DIR}/upgrade" -name "*.sql" ! -name "*_rollback.sql" | sort) else # 执行指定版本的变更 CHANGE_SCRIPTS=$(find "${DEPLOY_SQL_DIR}/upgrade/${VERSION}" -name "*.sql" ! -name "*_rollback.sql" | sort) fi ``` # 3. 执行变更脚本 echo "2. 执行变更脚本..." for script in ${CHANGE_SCRIPTS}; do echo "执行: $(basename ${script})" # 记录执行日志 mysql -h${DB_HOST} -u${DB_USER} -p${DB_PASSWORD} ${DB_NAME} < "${script}" if [ $? -eq 0 ]; then echo "✓ 执行成功" # 记录到变更历史表 echo "INSERT INTO db_changelog (script_name, execute_time, environment) VALUES ('$(basename ${script})', NOW(), '${ENVIRONMENT}');" | \ mysql -h${DB_HOST} -u${DB_USER} -p${DB_PASSWORD} ${DB_NAME} else echo "✗ 执行失败" exit 1 fi done # 4. 验证变更 echo "" echo "3. 验证变更结果..." # 执行验证脚本 VERIFY_SCRIPT="${DEPLOY_SQL_DIR}/upgrade/${VERSION}/verify_changes.sql" if [ -f "${VERIFY_SCRIPT}" ]; then mysql -h${DB_HOST} -u${DB_USER} -p${DB_PASSWORD} ${DB_NAME} < "${VERIFY_SCRIPT}" fi echo "数据库变更部署完成!" ``` ### 变更风险评估模板 ```markdown # 数据库变更风险评估表 ## 基本信息 - **变更编号**: DB-20240115-001 - **变更类型**: 字段添加/修改/删除 - **影响表**: user - **预计执行时间**: 5分钟 - **回滚时间**: 2分钟 ## 影响评估 ### 业务影响 - [ ] 用户注册/登录功能 - [ ] 数据查询性能 - [ ] 报表统计准确性 - [ ] 第三方系统对接 ### 技术风险 - [ ] 数据一致性风险 - [ ] 性能下降风险 - [ ] 锁表时间过长 - [ ] 空间不足风险 ## 预防措施 - [ ] 执行前完整备份 - [ ] 低峰期执行变更 - [ ] 准备回滚方案 - [ ] 安排专人监控 ## 应急预案 - **回滚步骤**: [详细回滚操作步骤] - **联系人员**: [紧急联系人列表] - **监控指标**: [关键监控项] ``` 这份参考手册提供了完整的实体类与数据库脚本同步的技术细节、工具脚本和生产环境部署规范。