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)
完整實現參考
推薦以下開源項目:
- bplustree - 純Python實現
- pybtree - 帶磁盤存儲支持
- 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