个人网站的设计论文,全国哪个餐饮品牌的网站做的好,网络推广预算方案,北京网站改版价格在设计 MMORPG#xff08;大规模多人在线角色扮演游戏#xff09;时#xff0c;数据库系统是游戏架构中至关重要的一部分。数据库不仅承担了游戏中各种数据#xff08;如玩家数据、物品数据、游戏世界状态等#xff09;的存储和管理任务#xff0c;还必须高效地支持并发访…在设计 MMORPG大规模多人在线角色扮演游戏时数据库系统是游戏架构中至关重要的一部分。数据库不仅承担了游戏中各种数据如玩家数据、物品数据、游戏世界状态等的存储和管理任务还必须高效地支持并发访问、事务处理和复杂的查询。为了确保系统的可扩展性和维护性我们需要对数据库操作进行封装和模块化设计。
为了实现上述目标本文设计了一个基于 Twisted 的数据库封装系统。Twisted 是一个异步框架适用于处理大量并发任务。结合 Twisted 和数据库连接池adbapi.ConnectionPool我们可以高效地执行异步数据库操作。 1. DatabaseError
作用自定义异常类用于在数据库操作发生错误时抛出详细的错误信息包含错误消息和错误码默认为500。2. DatabaseOperation
作用抽象基类定义了数据库操作的统一接口。所有具体的数据库操作类如 SelectOperation, InsertOperation 等都继承自此类必须实现 excute 方法来执行数据库事务。
关键方法
executeQuery执行实际的 SQL 查询。
handleSuccess操作成功时的回调函数。
handleFailure操作失败时的回调函数。3. SelectOperation
作用继承自 DatabaseOperation封装了 SELECT 查询操作。提供了 executeQuery 方法来执行 SQL 查询并返回查询结果。
关键方法
executeQuery执行 SELECT 查询并根据提供的表名、列名、查询条件等生成 SQL 语句。
excute实现 DatabaseOperation 中的抽象方法执行查询操作。4. InsertOperation
作用继承自 DatabaseOperation封装了 INSERT 插入操作。通过 executeQuery 方法生成插入的 SQL 语句并执行。
关键方法
executeQuery构建并执行 INSERT SQL 语句将数据插入指定的表。
excute实现 DatabaseOperation 中的抽象方法执行插入操作。5. UpdateOperation
作用继承自 DatabaseOperation封装了 UPDATE 更新操作。通过 executeQuery 方法生成更新的 SQL 语句并执行。
关键方法
executeQuery构建并执行 UPDATE SQL 语句用新值更新指定的行。
excute实现 DatabaseOperation 中的抽象方法执行更新操作。6. DeleteOperation
作用继承自 DatabaseOperation封装了 DELETE 删除操作。通过 executeQuery 方法生成删除的 SQL 语句并执行。
关键方法
executeQuery构建并执行 DELETE SQL 语句根据指定条件删除记录。
excute实现 DatabaseOperation 中的抽象方法执行删除操作。
7. DatabaseManager
作用负责数据库连接池的管理和数据库操作的执行。它使用 adbapi.ConnectionPool 创建数据库连接池执行操作并处理事务。
关键方法
getConnection返回数据库连接池的实例。
executeOperation接受一个数据库操作对象调用 runInteraction 方法来执行异步数据库事务并处理操作成功或失败的回调。
8. GameQueryPlayerId
作用继承自 SelectOperation封装了查询玩家信息的操作。它指定查询条件为玩家的名称并通过 excute 方法执行查询。
关键方法
excute执行 SelectOperation 中的 executeQuery 方法查询玩家 ID。
set_query_name设置查询的玩家名称。代码
from twisted.enterprise import adbapi
from twisted.internet.defer import Deferred
import pymysql
import traceback
from twisted.internet import reactor
from functools import partial
from abc import ABC, abstractmethod
# 异常类定义
class DatabaseError(Exception):def __init__(self, message, code500):self.message messageself.code code# 抽象的数据库操作类
class IDatabaseOperation(ABC):abstractmethoddef executeQuery(self, txn, table: str, columns: list, values: dict {}, condition: dict None) - any:passdef handleSuccess(self, result: any):print(Operation succeeded with result:, result)def handleFailure(self, error: Exception):# 这里做一些额外的错误处理比如记录日志或者返回友好的错误信息print(Operation failed:, error)abstractmethoddef excute(self, txn):#这里封装代码passclass ABC_SelectOperation(IDatabaseOperation):def executeQuery(self, txn, table: str, columns: list None, values: dict None, condition: dict None) - any:try:column_str , .join(columns) if columns else *query fSELECT {column_str} FROM {table}if condition:condition_str AND .join([f{key} %s for key in condition.keys()])query f WHERE {condition_str}print(fExecuting query: {query}, with values: {tuple(condition.values()) if condition else ()})txn.execute(query, tuple(condition.values()) if condition else ()) # use condition values if anyresult txn.fetchall()return resultexcept Exception as e:print(fError executing query: {e})traceback.print_exc()# 返回一个失败的结果以便继续后续操作return {error: str(e)}abstractmethoddef excute(self, txn):passclass ABC_InsertOperation(IDatabaseOperation):def executeQuery(self, txn, table: str, columns: list None, values: dict None, condition: dict None) - any:try:column_str , .join(columns)placeholders , .join([%s] * len(columns)) # Create placeholders based on column length# Extract values from the dictionary for each columnvalue_tuple tuple(values[col] for col in columns)query fINSERT INTO {table} ({column_str}) VALUES ({placeholders})print(fExecuting insert query: {query}, with values: {value_tuple})txn.execute(query, value_tuple) # Use parameterized queryreturn txn.lastrowid # Return the inserted record IDexcept Exception as e:print(fError executing insert query: {e})traceback.print_exc()# 返回一个失败的结果以便继续后续操作return {error: str(e)}abstractmethoddef excute(self, txn):passclass ABC_UpdateOperation(IDatabaseOperation):def executeQuery(self, txn, table: str, columns: list None, values: dict None, condition: dict None) - any:try:if condition and not isinstance(condition, dict):raise TypeError(Condition must be a dictionary)set_str , .join([f{col} %s for col in columns])query fUPDATE {table} SET {set_str}# Ensure that values is passed as a tuple for updatevalue_tuple tuple(values[col] for col in columns)if condition:condition_str AND .join([f{key} %s for key in condition.keys()])query f WHERE {condition_str}value_tuple tuple(condition.values()) # Append condition valuesprint(fExecuting update query: {query}, with values: {value_tuple})txn.execute(query, value_tuple) # Use parameterized querytxn.connection.commit()return txn.rowcount # Return the number of updated rowsexcept Exception as e:print(fError executing update query: {e})traceback.print_exc()# 返回一个失败的结果以便继续后续操作return {error: str(e)}abstractmethoddef excute(self, txn):passclass ABC_DeleteOperation(IDatabaseOperation):def executeQuery(self, txn, table: str, columns: list None, values: dict None, condition: dict None) - any:try:if not condition:raise ValueError(Condition for deletion cannot be empty.)condition_str AND .join([f{key} %s for key in condition.keys()])query fDELETE FROM {table} WHERE {condition_str}print(fExecuting delete query: {query}, with values: {tuple(condition.values())})txn.execute(query, tuple(condition.values())) # Use condition values for parameterized queryreturn txn.rowcountexcept Exception as e:print(fError executing delete query: {e})traceback.print_exc()# 返回一个失败的结果以便继续后续操作return {error: str(e)}abstractmethoddef excute(self):pass# 数据库管理类负责数据库连接池和事务
class DatabaseManager:def __init__(self, db_config):# 初始化数据库连接池self.dbConnectionPool adbapi.ConnectionPool(pymysql, **db_config)def getConnection(self):return self.dbConnectionPooldef executeOperation(self, operation: DatabaseOperation) - Deferred:try:# 使用 partial 创建一个指定了参数的函数deferred self.dbConnectionPool.runInteraction(operation.excute)deferred.addCallback(operation.handleSuccess)deferred.addErrback(operation.handleFailure)return deferredexcept Exception as e:error DatabaseError(str(e), 500)operation.handleFailure(error)return None# 示例数据库配置
db_config {host: localhost,user: root,password: root,database: test,
}class GameQueryPlayerId(ABC_SelectOperation):def __init__(self):self.select_columns [name, id]self.elect_condition {name: new_name} # 在此给出查询条件def excute(self, txn):# 执行查询操作return self.executeQuery(txn,test, columnsself.select_columns, conditionself.elect_condition )def set_query_name(self, name):self.elect_condition[name] name# 示例操作
def main():# 创建DatabaseManager实例db_manager DatabaseManager(db_config)ccGameQueryPlayerId GameQueryPlayerId()ccGameQueryPlayerId.set_query_name(new_name)deferred db_manager.executeOperation(ccGameQueryPlayerId)reactor.run()if __name__ __main__:main()
