## 1. 基本结构
$ cat > backup_tidb.sh << EOF #!/bin/bash # backup_tidb.sh # from:www.itpux.com.qq113257174.wx:itpux-com # web: http://www.fgedu.net.cn # 配置参数 BACKUP_DIR="/tidb/backup" DATE=$(date +%Y%m%d) BACKUP_NAME="tidb-backup-$DATE" # 创建备份目录 mkdir -p $BACKUP_DIR # 执行备份 echo "开始备份TiDB数据..." tiup br backup full --pd "192.168.1.100:2379" --storage "local://$BACKUP_DIR/$BACKUP_NAME" # 检查备份结果 if [ $? -eq 0 ]; then echo "备份成功：$BACKUP_DIR/$BACKUP_NAME" else echo "备份失败" exit 1 fi # 清理过期备份 find $BACKUP_DIR -name "tidb-backup-*" -mtime +7 -delete echo "清理过期备份完成" EOF ## 2. 执行脚本 $ chmod +x backup_tidb.sh $ ./backup_tidb.sh ## 3. 输出示例 开始备份TiDB数据... [2024/01/01 00:00:00] [INFO] [backup.go:268] ["full backup start"] [pd=192.168.1.100:2379] [storage=local:///tidb/backup/tidb-backup-20240101] [2024/01/01 00:00:05] [INFO] [backup.go:320] ["full backup success"] [backup-checksum=c8a1b2c3d4e5f6] 备份成功：/tidb/backup/tidb-backup-20240101 清理过期备份完成 ## 4. 定时执行 $ crontab -e 0 0 * * * /tidb/scripts/backup_tidb.sh

## 1. 基本结构
$ cat > data_migration.py << EOF #!/usr/bin/env python3 # data_migration.py # from:www.itpux.com.qq113257174.wx:itpux-com # web: http://www.fgedu.net.cn import pymysql import time def connect_db(host, port, user, password, db): """连接数据库""" try: conn = pymysql.connect( host=host, port=port, user=user, password=password, db=db, charset='utf8mb4' ) return conn except Exception as e: print(f"连接数据库失败: {e}") return None def migrate_data(src_conn, dst_conn, table_name, batch_size=1000): """迁移数据""" try: src_cursor = src_conn.cursor() dst_cursor = dst_conn.cursor() # 获取表结构 src_cursor.execute(f"SHOW CREATE TABLE {table_name}") create_table_sql = src_cursor.fetchone()[1] # 在目标数据库创建表 dst_cursor.execute(create_table_sql) dst_conn.commit() print(f"创建表 {table_name} 成功") # 获取数据总数 src_cursor.execute(f"SELECT COUNT(*) FROM {table_name}") total_rows = src_cursor.fetchone()[0] print(f"表 {table_name} 共有 {total_rows} 条数据") # 分批迁移数据 offset = 0 while offset < total_rows: # 查询数据 src_cursor.execute(f"SELECT * FROM {table_name} LIMIT {batch_size} OFFSET {offset}") rows = src_cursor.fetchall() if not rows: break # 构造插入语句 placeholders = ','.join(['%s'] * len(rows[0])) columns = ','.join([desc[0] for desc in src_cursor.description]) insert_sql = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})" # 批量插入数据 dst_cursor.executemany(insert_sql, rows) dst_conn.commit() offset += len(rows) print(f"已迁移 {offset}/{total_rows} 条数据") time.sleep(0.5) # 避免压力过大 print(f"表 {table_name} 数据迁移完成") except Exception as e: print(f"数据迁移失败: {e}") dst_conn.rollback() finally: if src_cursor: src_cursor.close() if dst_cursor: dst_cursor.close() def main(): # 源数据库连接 src_conn = connect_db('192.168.1.100', 4000, 'fgedu', 'password', 'fgedudb') if not src_conn: return # 目标数据库连接 dst_conn = connect_db('192.168.1.101', 4000, 'fgedu', 'password', 'fgedudb') if not dst_conn: src_conn.close() return # 迁移表数据 tables = ['fgedu_users', 'fgedu_orders', 'fgedu_payments'] for table in tables: print(f"开始迁移表 {table}") migrate_data(src_conn, dst_conn, table) # 关闭连接 src_conn.close() dst_conn.close() print("数据迁移完成") if __name__ == "__main__": main() EOF ## 2. 执行脚本 $ chmod +x data_migration.py $ python3 data_migration.py ## 3. 输出示例 开始迁移表 fgedu_users 创建表 fgedu_users 成功 表 fgedu_users 共有 5000000 条数据 已迁移 1000/5000000 条数据 已迁移 2000/5000000 条数据 ... 已迁移 5000000/5000000 条数据 表 fgedu_users 数据迁移完成 开始迁移表 fgedu_orders 创建表 fgedu_orders 成功 表 fgedu_orders 共有 10000000 条数据 已迁移 1000/10000000 条数据 已迁移 2000/10000000 条数据 ... 已迁移 10000000/10000000 条数据 表 fgedu_orders 数据迁移完成 开始迁移表 fgedu_payments 创建表 fgedu_payments 成功 表 fgedu_payments 共有 5000000 条数据 已迁移 1000/5000000 条数据 已迁移 2000/5000000 条数据 ... 已迁移 5000000/5000000 条数据 表 fgedu_payments 数据迁移完成 数据迁移完成

## 1. SQL批量插入
$ mysql -h 192.168.1.100 -P 4000 -u fgedu -p -e “INSERT INTO fgedudb.fgedu_users (id, name, email) VALUES (1, ‘user1’, ‘user1@fgedu.net.cn’), (2, ‘user2’, ‘user2@fgedu.net.cn’), (3, ‘user3’, ‘user3@fgedu.net.cn’);”

## 2. 输出示例
Query OK, 3 rows affected (0.01 sec)
Records: 3 Duplicates: 0 Warnings: 0

## 3. Python批量插入
$ cat > batch_insert.py << EOF #!/usr/bin/env python3 # batch_insert.py # from:www.itpux.com.qq113257174.wx:itpux-com # web: http://www.fgedu.net.cn import pymysql def batch_insert(): conn = pymysql.connect( host="192.168.1.100", port=4000, user="fgedu", password="password", db="fgedudb" ) cursor = conn.cursor() # 准备数据 data = [] for i in range(1000): data.append((i+1, f"user{i+1}", f"user{i+1}@fgedu.net.cn")) # 批量插入 sql = "INSERT INTO fgedu_users (id, name, email) VALUES (%s, %s, %s)" cursor.executemany(sql, data) conn.commit() print(f"批量插入 {cursor.rowcount} 条数据成功") cursor.close() conn.close() if __name__ == "__main__": batch_insert() EOF ## 4. 执行脚本 $ python3 batch_insert.py ## 5. 输出示例 批量插入 1000 条数据成功 ## 6. 分批批量插入 $ cat > batch_insert_batched.py << EOF #!/usr/bin/env python3 # batch_insert_batched.py # from:www.itpux.com.qq113257174.wx:itpux-com # web: http://www.fgedu.net.cn import pymysql def batch_insert_batched(): conn = pymysql.connect( host="192.168.1.100", port=4000, user="fgedu", password="password", db="fgedudb" ) cursor = conn.cursor() # 准备数据 total_rows = 10000 batch_size = 1000 for batch_start in range(0, total_rows, batch_size): data = [] batch_end = min(batch_start + batch_size, total_rows) for i in range(batch_start, batch_end): data.append((i+1, f"user{i+1}", f"user{i+1}@fgedu.net.cn")) # 批量插入 sql = "INSERT INTO fgedu_users (id, name, email) VALUES (%s, %s, %s)" cursor.executemany(sql, data) conn.commit() print(f"已插入 {batch_end}/{total_rows} 条数据") print(f"批量插入 {total_rows} 条数据成功") cursor.close() conn.close() if __name__ == "__main__": batch_insert_batched() EOF ## 7. 执行脚本 $ python3 batch_insert_batched.py ## 8. 输出示例 已插入 1000/10000 条数据 已插入 2000/10000 条数据 ... 已插入 10000/10000 条数据 批量插入 10000 条数据成功

## 1. SQL批量更新
$ mysql -h 192.168.1.100 -P 4000 -u fgedu -p -e “UPDATE fgedudb.fgedu_users SET email = CONCAT(name, ‘@fgedu.net.cn’) WHERE id IN (1, 2, 3);”

## 2. 输出示例
Query OK, 3 rows affected (0.01 sec)
Rows matched: 3 Changed: 3 Warnings: 0

## 3. Python批量更新
$ cat > batch_update.py << EOF #!/usr/bin/env python3 # batch_update.py # from:www.itpux.com.qq113257174.wx:itpux-com # web: http://www.fgedu.net.cn import pymysql def batch_update(): conn = pymysql.connect( host="192.168.1.100", port=4000, user="fgedu", password="password", db="fgedudb" ) cursor = conn.cursor() # 准备数据 data = [] for i in range(1000): data.append((f"user{i+1}@fgedu.net.cn", i+1)) # 批量更新 sql = "UPDATE fgedu_users SET email = %s WHERE id = %s" cursor.executemany(sql, data) conn.commit() print(f"批量更新 {cursor.rowcount} 条数据成功") cursor.close() conn.close() if __name__ == "__main__": batch_update() EOF ## 4. 执行脚本 $ python3 batch_update.py ## 5. 输出示例 批量更新 1000 条数据成功 ## 6. 分批批量更新 $ cat > batch_update_batched.py << EOF #!/usr/bin/env python3 # batch_update_batched.py # from:www.itpux.com.qq113257174.wx:itpux-com # web: http://www.fgedu.net.cn import pymysql def batch_update_batched(): conn = pymysql.connect( host="192.168.1.100", port=4000, user="fgedu", password="password", db="fgedudb" ) cursor = conn.cursor() # 准备数据 total_rows = 10000 batch_size = 1000 for batch_start in range(0, total_rows, batch_size): data = [] batch_end = min(batch_start + batch_size, total_rows) for i in range(batch_start, batch_end): data.append((f"user{i+1}@fgedu.net.cn", i+1)) # 批量更新 sql = "UPDATE fgedu_users SET email = %s WHERE id = %s" cursor.executemany(sql, data) conn.commit() print(f"已更新 {batch_end}/{total_rows} 条数据") print(f"批量更新 {total_rows} 条数据成功") cursor.close() conn.close() if __name__ == "__main__": batch_update_batched() EOF ## 7. 执行脚本 $ python3 batch_update_batched.py ## 8. 输出示例 已更新 1000/10000 条数据 已更新 2000/10000 条数据 ... 已更新 10000/10000 条数据 批量更新 10000 条数据成功

## 1. 编辑器
– Vim：强大的命令行编辑器
– VS Code：现代化的代码编辑器
– PyCharm：Python专用IDE
– Sublime Text：轻量级代码编辑器

## 2. 版本控制
– Git：分布式版本控制系统
– SVN：集中式版本控制系统

## 3. 调试工具
– pdb：Python调试器
– bashdb：Bash调试器
– strace：系统调用跟踪工具
– lsof：查看打开的文件

## 4. 测试工具
– pytest：Python测试框架
– unittest：Python内置测试框架
– bashunit：Bash测试框架
– shellcheck：Shell脚本检查工具

## 5. 性能分析工具
– cProfile：Python性能分析器
– time：命令执行时间测量
– top：系统资源监控
– iostat：I/O性能监控

## 6. 代码质量工具
– flake8：Python代码风格检查
– pylint：Python代码质量检查
– shellcheck：Shell脚本检查
– mccabe：代码复杂度分析

## 7. 依赖管理
– pip：Python包管理
– virtualenv：Python虚拟环境
– conda：Python环境管理
– npm：Node.js包管理

## 8. 文档工具
– Sphinx：Python文档生成器
– JSDoc：JavaScript文档生成器
– Doxygen：C++文档生成器
– MkDocs：Markdown文档生成器

## 9. 容器工具
– Docker：容器化平台
– Docker Compose：容器编排工具
– Kubernetes：容器编排平台

## 10. 自动化工具
– Jenkins：持续集成工具
– GitLab CI：持续集成工具
– Ansible：配置管理工具
– Terraform：基础设施即代码工具

## 1. 代码风格
– 遵循PEP 8：Python代码风格指南
– 遵循Shellcheck：Shell脚本风格指南
– 使用一致的缩进：4个空格或制表符
– 合理的命名：使用有意义的变量和函数名

## 2. 错误处理
– 捕获异常：使用try-except捕获异常
– 错误日志：记录错误信息
– 错误返回：使用非零退出码表示错误
– 错误处理：合理处理错误情况

## 3. 日志管理
– 日志级别：设置不同的日志级别
– 日志格式：统一日志格式
– 日志存储：存储日志文件
– 日志轮转：定期轮转日志文件

## 4. 配置管理
– 配置文件：使用配置文件管理配置
– 命令行参数：使用命令行参数覆盖配置
– 环境变量：使用环境变量传递敏感信息
– 默认值：设置合理的默认值

## 5. 安全性
– 输入验证：验证用户输入
– 密码处理：避免硬编码密码
– 权限控制：设置合理的文件权限
– 安全审计：记录脚本执行过程

## 6. 性能优化
– 批量操作：使用批量操作减少网络往返
– 并行处理：使用并行处理提高效率
– 内存管理：避免内存泄漏
– 资源释放：及时释放资源

## 7. 可维护性
– 模块化：将脚本分解为多个模块
– 文档：编写清晰的文档
– 注释：添加必要的注释
– 测试：编写测试用例

## 8. 版本控制
– 使用Git：使用Git管理代码
– 提交消息：编写清晰的提交消息
– 分支管理：使用分支管理不同版本
– 标签：使用标签标记重要版本

## 9. 部署
– 打包：打包脚本和依赖
– 安装：提供安装脚本
– 配置：提供配置模板
– 文档：提供部署文档

## 10. 监控
– 执行时间：监控脚本执行时间
– 资源使用：监控脚本资源使用
– 错误率：监控脚本错误率
– 输出：监控脚本输出