twisted实现MMORPG 游戏数据库操作封装设计与实现

在设计 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, code=500):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 = f"SELECT {column_str} FROM {table}"if condition:condition_str = " AND ".join([f"{key} = %s" for key in condition.keys()])query += f" WHERE {condition_str}"print(f"Executing 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(f"Error 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 = f"INSERT INTO {table} ({column_str}) VALUES ({placeholders})"print(f"Executing 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(f"Error 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 = f"UPDATE {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(f"Executing 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(f"Error 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 = f"DELETE FROM {table} WHERE {condition_str}"print(f"Executing 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(f"Error 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", columns=self.select_columns, condition=self.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()

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/18196.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

lvsDR模式实现

lvsDR模式实现

LVS DR模式与NAT模式的优缺点： NAT：优点：配置简单，所需网络环境简单，访问流量的出入都经过LVS服务器，控制流量简单， 缺点：由于访问流量的出入都会经过LVS服务器，所以LVS…
阅读更多...
闭源大语言模型的怎么增强:提示工程 检索增强生成 智能体

闭源大语言模型的怎么增强:提示工程 检索增强生成 智能体

闭源大语言模型的怎么增强 提示工程 检索增强生成 智能体 核心原理 提示工程：通过设计和优化提示词，引导大语言模型进行上下文学习和分解式思考，激发模型自身的思维和推理能力，使模型更好地理解和生成文本，增强其泛用性和解决问题的能力。检索增强生成：结合检索的准确…
阅读更多...
【快速幂算法】快速幂算法讲解及C语言实现（递归实现和非递归实现，附代码）

【快速幂算法】快速幂算法讲解及C语言实现（递归实现和非递归实现，附代码）

快速幂算法 快速幂算法可用分治法实现 不难看出，对任意实数a和非负整数n，有： a n { 1 , n 0 , a ≠ 0 0 , a 0 ( a n 2 ) 2 , n > 0 , n 为偶数 ( a n 2 ) 2 ∗ a , n > 0 , n 为奇数 a^n \begin{cases} 1, & n 0, a\neq 0…
阅读更多...
大数据SQL调优专题——Hive执行原理

大数据SQL调优专题——Hive执行原理

引入 Apache Hive 是基于Hadoop的数据仓库工具，它可以使用SQL来读取、写入和管理存在分布式文件系统中的海量数据。在Hive中，HQL默认转换成MapReduce程序运行到Yarn集群中，大大降低了非Java开发者数据分析的门槛，并且Hive提供命令…
阅读更多...
30天开发操作系统 第 20 天 -- API

30天开发操作系统 第 20 天 -- API

前言 大家早上好，今天我们继续努力哦。 昨天我们已经实现了应用程序的运行, 今天我们来实现由应用程序对操作系统功能的调用(即API, 也叫系统调用)。 为什么这样的功能称为“系统调用”(system call)呢？因为它是由应用程序来调用(操作)系统中的功能来完…
阅读更多...
UE5.2后 Bake Out Materials失效

UE5.2后 Bake Out Materials失效

这个问题出现在5.3，5.4，5.5没有测试 烘焙贴图后会找不到贴图位置， 这个是5.2的正常状态 默认是生成在模型当前目录里，包括新的材质 但是这个bug会让材质和贴图都消失，无法定位 暂时没有办法解决，等官方 …
阅读更多...
macOS部署DeepSeek-r1

macOS部署DeepSeek-r1

好奇，跟着网友们的操作试了一下 网上方案很多，主要参考的是这篇 DeepSeek 接入 PyCharm，轻松助力编程_pycharm deepseek-CSDN博客 方案是：PyCharm CodeGPT插件 DeepSeek-r1:1.5b 假设已经安装好了PyCharm PyCharm: the Pyth…
阅读更多...
架构设计系列（二）：CI/CD

架构设计系列（二）：CI/CD

一、概述 CI/CD 是 持续集成（Continuous Integration） 和 持续交付/持续部署（Continuous Delivery/Continuous Deployment） 的缩写，是现代软件开发中的一套核心实践和工具链，旨在提高软件交付的效率、质量…
阅读更多...
Windows 11 搭建私有知识库（docker、dify、deepseek、ollama）

Windows 11 搭建私有知识库（docker、dify、deepseek、ollama）

一、操作系统信息 版本 Windows 11 家庭中文版 版本号 23H2 安装日期 ‎2023/‎8/‎21 操作系统版本 22631.4460二、搭建思路 ollama拉取deepseek、bge-m3模型docker拉取dify的镜像dify链接ollama使用模型，并上传文件搭建知识库，创建应用 三、搭建步骤…
阅读更多...
本地部署DeepSeek摆脱服务器繁忙

本地部署DeepSeek摆脱服务器繁忙

由于图片和格式解析问题，可前往 阅读原文 最近DeepSeek简直太火了，频频霸榜热搜打破春节的平静，大模型直接开源让全球科技圈都为之震撼！再次证明了中国AI的换道超车与崛起 DeepSeek已经成了全民ai，使用量也迅速上去了…
阅读更多...
‌CBA认证‌（业务架构师认证）简介---适用人群、考试内容与形式、含金量与职业前景，以及‌CBA、TOGAF认证对比表格

‌CBA认证‌（业务架构师认证）简介---适用人群、考试内容与形式、含金量与职业前景，以及‌CBA、TOGAF认证对比表格

‌CBA认证‌，即业务架构师认证（Certified Business Architect，CBA），是由业务架构师协会（Business Architecture Institute）推出的一项国际认证计划。该认证旨在评估和认证业务架构师的专业能力和…
阅读更多...
保姆级GitHub大文件（100mb-2gb）上传教程

保姆级GitHub大文件（100mb-2gb）上传教程

GLF（Git Large File Storage）安装使用 使用GitHub desktop上传大于100mb的文件时报错 The following files are over 100MB. lf you commit these files, you will no longer beable to push this repository to GitHub.com.term.rarWe recommend you a…
阅读更多...
使用 Visual Studio Code (VS Code) 开发 Python 图形界面程序

使用 Visual Studio Code (VS Code) 开发 Python 图形界面程序

安装Python、VS Code Documentation for Visual Studio Code Python Releases for Windows | Python.org 更新pip >python.exe -m pip install --upgrade pip Requirement already satisfied: pip in c:\users\xxx\appdata\local\programs\python\python312\lib\site-pa…
阅读更多...
Python的那些事第二十一篇：Python Web开发的“秘密武器”Flask

Python的那些事第二十一篇：Python Web开发的“秘密武器”Flask

基于 Flask 框架的 Python Web 开发研究 摘要 在 Web 开发的江湖里，Python 是一位武林高手，而 Flask 则是它手中那把小巧却锋利的匕首。本文以 Flask 框架为核心，深入探讨了它在 Python Web 开发中的应用。通过幽默风趣的笔触，结合实例和表格，分析了 Flask 的特性、优势以…
阅读更多...
Qt开发①Qt的概念+发展+优点+应用+使用

Qt开发①Qt的概念+发展+优点+应用+使用

目录 1. Qt的概念和发展 1.1 Qt的概念 1.2 Qt 的发展史： 1.3 Qt 的版本 2. Qt 的优点和应用 2.1 Qt 的优点： 2.2 Qt 的应用场景 2.3 Qt 的应用案例 3. 搭建 Qt 开发环境 3.1 Qt 的开发工具 3.2 Qt SDK 的下载和安装 3.3 Qt 环境变量配置和使…
阅读更多...
【第4章：循环神经网络（RNN）与长短时记忆网络（LSTM）— 4.3 RNN与LSTM在自然语言处理中的应用案例】

【第4章：循环神经网络（RNN）与长短时记忆网络（LSTM）— 4.3 RNN与LSTM在自然语言处理中的应用案例】

咱今天来聊聊在人工智能领域里，特别重要的两个神经网络：循环神经网络（RNN）和长短时记忆网络（LSTM），主要讲讲它们在自然语言处理里的应用。你想想，平常咱们用手机和别人聊天、看新闻、听语音助手说话，背后说不定就有 RNN 和 LSTM 在帮忙呢！ 二、RNN 是什么？ （一）…
阅读更多...
DeepSeek应用——与PyCharm的配套使用

DeepSeek应用——与PyCharm的配套使用

目录 一、配置方法 二、使用方法 三、注意事项 1、插件市场无continue插件 2、无结果返回，且在本地模型报错 记录自己学习应用DeepSeek的过程，使用的是自己电脑本地部署的私有化蒸馏模型...... （举一反三，这个不单单是可以用…
阅读更多...
国自然地区基金｜影像组学联合病理组学预测进展期胃癌术后预后的研究｜基金申请·25-02-13

国自然地区基金｜影像组学联合病理组学预测进展期胃癌术后预后的研究｜基金申请·25-02-13

小罗碎碎念 今天和大家分享一个国自然地区科学项目，执行年限为2020.01～2023.12，直接费用为34万元。 胃癌在我国发病形势严峻，现有TNM分期预后评估存在局限，难以满足精准医疗需求。本项目运用“医工结合，学科…
阅读更多...
【Java集合一】集合概述

【Java集合一】集合概述

一、集合简介 Java 集合框架（Collection Framework）是 Java 提供的一组用于存储和操作对象的类和接口集合。这些集合类提供了不同的数据结构，使得数据的管理和操作更加方便和高效。 Java 集合框架提供了各种类型的数据结构，如列…
阅读更多...
k8s集群搭建参考（by lqw）

k8s集群搭建参考（by lqw）

文章目录 声明配置yum源安装docker安装 kubeadm，kubelet 和 kubectl部署主节点其他节点加入集群安装网络插件 声明 由于看了几个k8s的教程，都存在各种问题，自己搭建的时候，踩了不少坑，最后还是靠百度csdnchatGPT才搭建…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023