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 arr
    pivot = 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)-1
    while low <= high:
        mid = (low + high) // 2
        if arr[mid] == target:
            return mid
        elif arr[mid] < target:
            low = mid + 1
        else:
            high = mid - 1
    return -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 1
    memo[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] = 0
    heap = [(0, start)]
    while heap:
        current_dist, node = heapq.heappop(heap)
        for neighbor, weight in graph[node].items():
            distance = current_dist + weight
            if distance < distances[neighbor]:
                distances[neighbor] = distance
                heapq.heappush(heap, (distance, neighbor))
    return distances

字符串处理

KMP字符串匹配

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

    lps = build_lps(pattern)
    i = j = 0
    while i < len(text):
        if pattern[j] == text[i]:
            i += 1
            j += 1
            if j == len(pattern):
                return i - j
        else:
            if j != 0:
                j = lps[j-1]
            else:
                i += 1
    return -1

数学算法

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

def sieve_of_eratosthenes(limit):
    sieve = [True] * (limit+1)
    sieve[0] = sieve[1] = False
    for 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 = capacity

    def get(self, key):
        if key not in self.cache: return -1
        self.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] = value
        if 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_leaf
        self.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)
        return
    
    leaf = 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 results
        leaf = leaf.next_leaf
    return 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 Lock

class 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