Python算法实战：从排序到B+树全解析

Python中常见的算法示例

以下是Python中常见的算法示例，涵盖基础算法和经典问题解决方案，代码可直接运行：

排序算法

冒泡排序

def bubble_sort(arr):n = len(arr)for i in range(n):for j in range(0, n-i-1):if arr[j] > arr[j+1]:arr[j], arr[j+1] = arr[j+1], arr[j]

快速排序

def quick_sort(arr):if len(arr) <= 1:return arrpivot = arr[len(arr)//2]left = [x for x in arr if x < pivot]middle = [x for x in arr if x == pivot]right = [x for x in arr if x > pivot]return quick_sort(left) + middle + quick_sort(right)

搜索算法

二分查找

def binary_search(arr, target):low, high = 0, len(arr)-1while low <= high:mid = (low + high) // 2if arr[mid] == target:return midelif arr[mid] < target:low = mid + 1else:high = mid - 1return -1

广度优先搜索（BFS）

from collections import deque
def bfs(graph, start):visited = set()queue = deque([start])while queue:vertex = queue.popleft()if vertex not in visited:visited.add(vertex)queue.extend(graph[vertex] - visited)return visited

---

动态规划

斐波那契数列（带缓存）

def fib(n, memo={}):if n in memo: return memo[n]if n <= 2: return 1memo[n] = fib(n-1, memo) + fib(n-2, memo)return memo[n]

背包问题

def knapsack(W, wt, val, n):dp = [[0]*(W+1) for _ in range(n+1)]for i in range(1, n+1):for w in range(1, W+1):if wt[i-1] <= w:dp[i][w] = max(val[i-1] + dp[i-1][w-wt[i-1]], dp[i-1][w])else:dp[i][w] = dp[i-1][w]return dp[n][W]

图算法

Dijkstra最短路径

import heapq
def dijkstra(graph, start):distances = {node: float('inf') for node in graph}distances[start] = 0heap = [(0, start)]while heap:current_dist, node = heapq.heappop(heap)for neighbor, weight in graph[node].items():distance = current_dist + weightif distance < distances[neighbor]:distances[neighbor] = distanceheapq.heappush(heap, (distance, neighbor))return distances

字符串处理

KMP字符串匹配

def kmp_search(pattern, text):def build_lps(pattern):lps = [0] * len(pattern)length = 0for i in range(1, len(pattern)):while length > 0 and pattern[i] != pattern[length]:length = lps[length-1]if pattern[i] == pattern[length]:length += 1lps[i] = lengthreturn lpslps = build_lps(pattern)i = j = 0while i < len(text):if pattern[j] == text[i]:i += 1j += 1if j == len(pattern):return i - jelse:if j != 0:j = lps[j-1]else:i += 1return -1

数学算法

素数筛选（埃拉托斯特尼筛法）

def sieve_of_eratosthenes(limit):sieve = [True] * (limit+1)sieve[0] = sieve[1] = Falsefor num in range(2, int(limit**0.5)+1):if sieve[num]:sieve[num*num::num] = [False]*len(sieve[num*num::num])return [i for i, is_prime in enumerate(sieve) if is_prime]

其他经典算法

汉诺塔问题

def hanoi(n, source, target, auxiliary):if n > 0:hanoi(n-1, source, auxiliary, target)print(f"Move disk {n} from {source} to {target}")hanoi(n-1, auxiliary, target, source)

LRU缓存实现

from collections import OrderedDict
class LRUCache:def __init__(self, capacity):self.cache = OrderedDict()self.capacity = capacitydef get(self, key):if key not in self.cache: return -1self.cache.move_to_end(key)return self.cache[key]def put(self, key, value):if key in self.cache:self.cache.move_to_end(key)self.cache[key] = valueif len(self.cache) > self.capacity:self.cache.popitem(last=False)

以上示例覆盖了排序、搜索、动态规划、图算法等核心领域，可根据实际需求调整参数和数据结构。每个算法均经过简化以突出核心逻辑，实际应用时需考虑边界条件和异常处理。

于Python中B+树

以下是关于Python中B+树实现的实用示例，涵盖基础操作、高级应用及优化场景。内容基于开源项目、算法书籍和实际工程案例整理，避免理论赘述，直接提供可运行的代码片段。

基础实现类

class BPlusTreeNode:def __init__(self, is_leaf=False):self.keys = []self.children = []self.is_leaf = is_leafself.next_leaf = None  # 用于叶子节点连接

插入操作

def insert(self, key, value):if not self.root:self.root = BPlusTreeNode(is_leaf=True)self.root.keys.append(key)self.root.children.append(value)returnleaf = self._find_leaf(key)self._insert_into_leaf(leaf, key, value)if len(leaf.keys) > self.order:self._split_leaf(leaf)

范围查询

def range_query(self, start, end):results = []leaf = self._find_leaf(start)while leaf:for i, key in enumerate(leaf.keys):if start <= key <= end:results.append(leaf.children[i])elif key > end:return resultsleaf = leaf.next_leafreturn results

磁盘持久化

def save_to_disk(self, filename):with open(filename, 'wb') as f:pickle.dump({'root': self.root,'order': self.order}, f)

实际应用场景示例

1. 数据库索引

class DatabaseIndex:def __init__(self):self.bptree = BPlusTree(order=50)def add_record(self, pk, record):self.bptree.insert(pk, record)

2. 时间序列数据存储

def store_time_series(timestamps, values):tree = BPlusTree(order=100)for ts, val in zip(timestamps, values):tree.insert(ts, val)return tree

3. 文件系统目录

class FileSystemDir:def __init__(self):self.tree = BPlusTree(order=32)def add_file(self, filename, inode):self.tree.insert(filename, inode)

---

高级优化技巧

批量加载（Bulk Loading）

def bulk_load(sorted_items):tree = BPlusTree()leaves = []for i in range(0, len(sorted_items), tree.order):leaf = BPlusTreeNode(is_leaf=True)leaf.keys = sorted_items[i:i+tree.order][0]leaf.children = sorted_items[i:i+tree.order][1]leaves.append(leaf)_build_non_leaf_levels(leaves)

并发控制

from threading import Lockclass ConcurrentBPlusTree:def __init__(self):self.lock = Lock()self.tree = BPlusTree()def safe_insert(self, key, value):with self.lock:self.tree.insert(key, value)

---

完整实现参考

推荐以下开源项目：

1. bplustree - 纯Python实现
2. pybtree - 带磁盘存储支持
3. BPlusTree - 教学级清晰实现

每个项目包含完整的插入、删除、搜索实现，以及性能测试案例。实际工程中建议直接使用这些成熟实现而非重复造轮子。

---

性能测试片段

import time
test_tree = BPlusTree(order=100)
start = time.time()
for i in range(100000):test_tree.insert(i, f"value_{i}")
print(f"Insert time: {time.time()-start:.2f}s")

特殊场景处理

变长键支持

def _compare_keys(a, b):if isinstance(a, str) or isinstance(b, str):return str(a) < str(b)return a < b

自定义序列化

def serialize_node(node):return {'is_leaf': node.is_leaf,'keys': pickle.dumps(node.keys),'children': pickle.dumps(node.children)}

以上示例覆盖了B+树在Python中的核心应用场景，可根据具体需求组合使用。实际开发时需注意：

• 节点大小与磁盘页对齐（通常4KB）
• 内存缓冲层设计
• 并发访问时的锁粒度控制

完整工程实现建议参考PostgreSQL或MySQL的B+树索引源码。

Python操作PostgreSQL的实用示例

以下是使用Python操作PostgreSQL的实用示例，涵盖连接、CRUD、事务管理等常见操作。假设已安装psycopg2库（可通过pip install psycopg2安装）。

---

基础连接与配置

连接PostgreSQL数据库

import psycopg2
conn = psycopg2.connect(host="localhost",database="testdb",user="postgres",password="yourpassword"
)
cursor = conn.cursor()

使用环境变量管理连接参数

import os
from psycopg2 import connect
conn = connect(host=os.getenv("DB_HOST"),database=os.getenv("DB_NAME"),user=os.getenv("DB_USER"),password=os.getenv("DB_PASS")
)

检查连接状态

print(conn.closed)  # 0表示连接正常

使用上下文管理器自动关闭连接

with psycopg2.connect(**params) as conn:with conn.cursor() as cursor:cursor.execute("SELECT version()")print(cursor.fetchone())

配置连接池（需psycopg2.pool）

from psycopg2 import pool
connection_pool = pool.SimpleConnectionPool(minconn=1,maxconn=10,**params
)
conn = connection_pool.getconn()

---

表操作

创建表

cursor.execute("""CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY,name VARCHAR(50) NOT NULL,email VARCHAR(100) UNIQUE)
""")
conn.commit()

删除表

cursor.execute("DROP TABLE IF EXISTS temp_data")
conn.commit()

添加列

cursor.execute("ALTER TABLE users ADD COLUMN age INTEGER")
conn.commit()

创建索引

cursor.execute("CREATE INDEX idx_user_email ON users(email)")
conn.commit()

查看所有表

cursor.execute("""SELECT table_name FROM information_schema.tables WHERE table_schema='public'
""")
print(cursor.fetchall())

---

CRUD操作

插入单条数据

cursor.execute("INSERT INTO users (name, email) VALUES (%s, %s)",("Alice", "alice@example.com")
)
conn.commit()

批量插入

data = [("Bob", "bob@test.com"), ("Charlie", "charlie@test.com")]
cursor.executemany("INSERT INTO users (name, email) VALUES (%s, %s)",data
)
conn.commit()

查询数据

cursor.execute("SELECT * FROM users WHERE name LIKE %s", ("A%",))
rows = cursor.fetchall()
for row in rows:print(row)

更新数据

cursor.execute("UPDATE users SET email = %s WHERE id = %s",("new_email@test.com", 1)
)
conn.commit()

删除数据

cursor.execute("DELETE FROM users WHERE id = %s", (5,))
conn.commit()

带返回值的INSERT

cursor.execute("""INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id
""", ("Dave", "dave@test.com"))
new_id = cursor.fetchone()[0]
print(f"New record ID: {new_id}")

---

高级查询

分页查询

page = 2
limit = 10
offset = (page - 1) * limit
cursor.execute("SELECT * FROM users ORDER BY id LIMIT %s OFFSET %s",(limit, offset)
)

使用JOIN查询

cursor.execute("""SELECT u.name, o.order_date FROM users u JOIN orders o ON u.id = o.user_id
""")

聚合函数

cursor.execute("""SELECT COUNT(*), AVG(age) FROM users WHERE age > 18
""")
count, avg_age = cursor.fetchone()

事务处理

try:cursor.execute("INSERT INTO table1 VALUES (%s)", (value1,))cursor.execute("UPDATE table2 SET col1 = %s", (value2,))conn.commit()
except:conn.rollback()

使用WITH子句（CTE）

cursor.execute("""WITH active_users AS (SELECT * FROM users WHERE last_login > NOW() - INTERVAL '30 days')SELECT COUNT(*) FROM active_users
""")

查看表结构

cursor.execute("""SELECT column_name, data_type FROM information_schema.columns WHERE table_name = 'users'
""")
print(cursor.fetchall())

数据类型处理

处理JSON数据

import json
data = {"key": "value"}
cursor.execute("INSERT INTO config (config_data) VALUES (%s)",(json.dumps(data),)
)

处理数组类型

cursor.execute("INSERT INTO products (tags) VALUES (%s)",(['electronics', 'gadgets'],)
)

处理日期时间

from datetime import datetime
cursor.execute("INSERT INTO events (event_name, event_date) VALUES (%s, %s)",("Meeting", datetime.now())
)

处理二进制数据

with open('image.png', 'rb') as f:cursor.execute("INSERT INTO images (name, data) VALUES (%s, %s)",("logo", f.read()))

---

性能优化

使用PREPARE语句

cursor.execute("PREPARE userplan AS SELECT * FROM users WHERE id = $1")
cursor.execute("EXECUTE userplan (1)")

大批量数据COPY

with open('data.csv', 'w') as f:cursor.copy_expert("COPY users TO STDOUT WITH CSV HEADER", f)

使用EXPLAIN分析查询

cursor.execute("EXPLAIN ANALYZE SELECT * FROM users")
print(cursor.fetchall())

设置超时参数

cursor.execute("SET statement_timeout TO 1000")  # 毫秒

监控长时间运行查询

cursor.execute("""SELECT pid, query, now() - query_start AS duration FROM pg_stat_activity WHERE state = 'active'
""")

维护与监控

备份数据库（需pg_dump）

import subprocess
subprocess.run(["pg_dump", "-U", "postgres", "-d", "dbname", "-f", "backup.sql"])

查看锁信息

cursor.execute("""SELECT locktype, relation::regclass, mode FROM pg_locks WHERE relation = 'users'::regclass
""")

查看数据库大小

cursor.execute("SELECT pg_size_pretty(pg_database_size(current_database()))")
print(cursor.fetchone())

清理数据库

cursor.execute("VACUUM FULL ANALYZE")
conn.commit()

查看扩展列表

cursor.execute("SELECT * FROM pg_available_extensions")
print(cursor.fetchall())

以上示例覆盖了PostgreSQL的常见使用场景。实际应用中需根据业务需求调整参数和安全措施（如使用参数化查询防止SQL注入）。

Python在AI中实例

Python在AI中的应用方法

Python是人工智能领域的主流语言，拥有丰富的库和框架支持。其简洁语法和强大生态使其成为开发机器学习、深度学习等AI项目的首选。

常用Python AI库与框架

TensorFlow与PyTorch

• TensorFlow由Google开发，适合大规模分布式训练和生产部署，提供Keras高层API简化开发
• PyTorch由Facebook维护，动态计算图更灵活，研究领域使用广泛

# PyTorch示例
import torch
model = torch.nn.Sequential(torch.nn.Linear(8, 32),torch.nn.ReLU(),torch.nn.Linear(32, 1)
)

Scikit-learn

• 提供经典机器学习算法实现
• 包含数据预处理、模型评估工具
• 适合中小规模结构化数据

from sklear