本文目录导读：
我来帮你实现一个Python命令查询模式下的读写分离功能,这里提供几种常见的实现方案：
方案1：基础读写分离装饰器
import functools
from enum import Enum
class Mode(Enum):
READ = "read"
WRITE = "write"
class ReadWriteSeparator:
def __init__(self):
self._current_mode = Mode.READ
@property
def current_mode(self):
return self._current_mode
@current_mode.setter
def current_mode(self, mode):
if mode not in Mode:
raise ValueError(f"Invalid mode: {mode}")
self._current_mode = mode
def read(self, func):
"""装饰读操作"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
if self._current_mode != Mode.READ:
raise RuntimeError("当前不是读模式")
return func(*args, **kwargs)
return wrapper
def write(self, func):
"""装饰写操作"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
if self._current_mode != Mode.WRITE:
raise RuntimeError("当前不是写模式")
return func(*args, **kwargs)
return wrapper
# 使用示例
separator = ReadWriteSeparator()
class DataManager:
def __init__(self):
self.data = {}
@separator.read
def get_data(self, key):
return self.data.get(key)
@separator.write
def set_data(self, key, value):
self.data[key] = value
@separator.read
def list_keys(self):
return list(self.data.keys())
# 测试
dm = DataManager()
separator.current_mode = Mode.WRITE
dm.set_data("name", "Alice") # 可以写入
separator.current_mode = Mode.READ
print(dm.get_data("name")) # 可以读取
try:
dm.set_data("age", 25) # 会抛出异常
except RuntimeError as e:
print(f"错误: {e}")
方案2：基于命令模式实现
from abc import ABC, abstractmethod
from typing import Any, Dict, List
class Command(ABC):
"""命令基类"""
@abstractmethod
def execute(self):
pass
@abstractmethod
def is_readonly(self) -> bool:
pass
class ReadCommand(Command):
"""读命令基类"""
def is_readonly(self):
return True
class WriteCommand(Command):
"""写命令基类"""
def is_readonly(self):
return False
class DataStore:
def __init__(self):
self._data: Dict[str, Any] = {}
self._read_commands = []
self._write_commands = []
def execute(self, command: Command):
"""执行命令"""
if command.is_readonly():
self._read_commands.append(command)
return command.execute()
else:
self._write_commands.append(command)
return command.execute()
def get_data(self, key: str) -> Any:
"""创建读命令"""
class GetCommand(ReadCommand):
def execute(self):
return self._data.get(key)
return self.execute(GetCommand())
def set_data(self, key: str, value: Any):
"""创建写命令"""
class SetCommand(WriteCommand):
def execute(self):
self._data[key] = value
return True
return self.execute(SetCommand())
def get_command_history(self) -> Dict[str, List]:
"""获取命令历史"""
return {
"read": self._read_commands,
"write": self._write_commands
}
# 使用示例
store = DataStore()
store.set_data("name", "Bob")
store.set_data("age", 30)
print(store.get_data("name")) # Bob
print(store.get_data("city")) # None
方案3：支持读写分离的数据库操作
import sqlite3
from contextlib import contextmanager
from typing import Optional
class DatabaseReadWriteSeparator:
"""数据库读写分离管理器"""
def __init__(self, read_db_path: str, write_db_path: str):
self._read_db_path = read_db_path
self._write_db_path = write_db_path
self._connections = {}
@contextmanager
def read_connection(self):
"""获取读连接"""
conn = sqlite3.connect(self._read_db_path)
self._connections['read'] = conn
try:
yield conn
finally:
conn.close()
self._connections.pop('read', None)
@contextmanager
def write_connection(self):
"""获取写连接"""
conn = sqlite3.connect(self._write_db_path)
self._connections['write'] = conn
try:
yield conn
finally:
conn.close()
self._connections.pop('write', None)
def execute_read(self, query: str, params: tuple = ()) -> Optional[list]:
"""执行读操作"""
with self.read_connection() as conn:
cursor = conn.cursor()
cursor.execute(query, params)
return cursor.fetchall()
def execute_write(self, query: str, params: tuple = ()) -> bool:
"""执行写操作"""
with self.write_connection() as conn:
cursor = conn.cursor()
cursor.execute(query, params)
conn.commit()
return True
def sync_databases(self):
"""同步读写数据库（示例功能）"""
# 这里可以实现主从同步逻辑
pass
# 使用示例
class UserRepository:
def __init__(self, db_separator: DatabaseReadWriteSeparator):
self.db = db_separator
def get_user(self, user_id: int):
"""读操作"""
query = "SELECT * FROM users WHERE id = ?"
return self.db.execute_read(query, (user_id,))
def create_user(self, name: str, email: str):
"""写操作"""
query = "INSERT INTO users (name, email) VALUES (?, ?)"
return self.db.execute_write(query, (name, email))
def update_user(self, user_id: int, name: str):
"""写操作"""
query = "UPDATE users SET name = ? WHERE id = ?"
return self.db.execute_write(query, (name, user_id))
方案4：基于SQLAlchemy的读写分离
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, Session
from typing import Optional
class SQLAlchemyReadWriteSeparator:
"""SQLAlchemy读写分离封装"""
def __init__(self, read_url: str, write_url: str):
self._read_engine = create_engine(read_url)
self._write_engine = create_engine(write_url)
self._ReadSession = sessionmaker(bind=self._read_engine)
self._WriteSession = sessionmaker(bind=self._write_engine)
@contextmanager
def read_session(self) -> Session:
"""获取读会话"""
session = self._ReadSession()
try:
yield session
finally:
session.close()
@contextmanager
def write_session(self) -> Session:
"""获取写会话"""
session = self._WriteSession()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
def execute_read_query(self, sql: str, params: dict = None) -> list:
"""执行读SQL"""
with self.read_session() as session:
result = session.execute(text(sql), params or {})
return result.fetchall()
def execute_write_query(self, sql: str, params: dict = None) -> int:
"""执行写SQL"""
with self.write_session() as session:
result = session.execute(text(sql), params or {})
return result.rowcount
使用建议
- 选择方案：根据你的具体需求选择合适的方案
- 异常处理：添加适当的异常处理机制
- 日志记录：记录读写操作日志以便调试
- 性能监控：监控读写分离的性能效果
- 数据一致性：确保读写分离后的数据一致性
需要我详细解释某个特定方案或提供更多示例吗？
