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.mzph.cn/news/895818.shtml

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

相关文章

【R语言】聚类分析

【R语言】聚类分析

聚类分析是一种常用的无监督学习方法，是将所观测的事物或者指标进行分类的一种统计分析方法，其目的是通过辨认在某些特征上相似的事物，并将它们分成各种类别。R语言提供了多种聚类分析的方法和包。 方法优点缺点适用场景K-means计算效率高需…
阅读更多...
超全Deepseek资料包，deepseek下载安装部署提示词及本地部署指南介绍

超全Deepseek资料包，deepseek下载安装部署提示词及本地部署指南介绍

该资料包涵盖了DeepSeek模型的下载、安装、部署以及本地运行的详细指南，适合希望在本地环境中高效运行DeepSeek模型的用户。资料包不仅包括基础的安装步骤，还提供了68G多套独立部署视频教程教程，针对不同硬件配置的模型选择建议，以…
阅读更多...
Java Spring boot 篇：常用注解

Java Spring boot 篇：常用注解

Configuration 作用 Configuration 注解的核心作用是把一个类标记为 Spring 应用上下文里的配置类。配置类就像一个 Java 版的 XML 配置文件，能够在其中定义 Bean 定义和 Bean 之间的依赖关系。当 Spring 容器启动时，会扫描这些配置类，解析其…
阅读更多...
在 Ubuntu 20.04 为 Clash Verge AppImage 创建桌面图标教程

在 Ubuntu 20.04 为 Clash Verge AppImage 创建桌面图标教程

在 Ubuntu 20.04 为 AppImage 创建桌面图标教程 一、准备工作 确保你已经下载了 xxxx.AppImage 文件，并且知道它所在的具体路径。同时，你可以准备一个合适的图标文件（.png 格式）用于代表该应用程序，如果没有合适的图…
阅读更多...
【复现DeepSeek-R1之Open R1实战】系列6：GRPO源码逐行深度解析（上）

【复现DeepSeek-R1之Open R1实战】系列6：GRPO源码逐行深度解析（上）

目录 4 GRPO源码分析4.1 数据类 GRPOScriptArguments4.2 系统提示字符串 SYSTEM_PROMPT4.3 奖励函数4.3.1 accuracy_reward函数4.3.2 verify函数4.3.3 format_reward函数 4.4 将数据集格式化为对话形式4.5 初始化GRPO Trainer 【复现DeepSeek-R1之Open R1实战】系列3&#xff1…
阅读更多...
【杂谈】加油！！！！

【杂谈】加油！！！！

为了在三月底前系统准备Java后端开发的面试和笔试，以下是分阶段的高效学习计划： 一、知识体系构建（第1-2周） 核心基础强化 Java基础（每日1.5小时）： 重点掌握：JVM内存模型&#xff0…
阅读更多...
python旅游推荐系统+爬虫+可视化（协同过滤算法）

python旅游推荐系统+爬虫+可视化（协同过滤算法）

✅️基于用户的协同过滤算法 ✅️有后台管理 ✅️2w多数据集 这个旅游数据分析推荐系统采用了Python语言、Django框架、MySQL数据库、requests库进行网络爬虫开发、机器学习中的协同过滤算法、ECharts数据可视化技术，以实现从网站抓取旅游数据、个性化推荐和直观展…
阅读更多...
HarmonyNext上传用户相册图片到服务器

HarmonyNext上传用户相册图片到服务器

图片选择就不用说了，直接用 无须申请权限 。 上传图片，步骤和android对比稍微有点复杂，可能是为了安全性考虑，需要将图片先拷贝到缓存目录下面，然后再上传，当然你也可以转成Base64，然后和服务…
阅读更多...
同为科技智能PDU助力Deepseek人工智能和数据交互的快速发展

同为科技智能PDU助力Deepseek人工智能和数据交互的快速发展

1 2025开年，人工智能领域迎来了一场前所未有的变革。Deepseek成为代表“东方力量”的开年王炸，不仅在国内掀起了技术热潮，并且在全球范围内引起了高度关注。Deepseek以颠覆性技术突破和现象级应用场景席卷全球，这不仅重塑了产业格…
阅读更多...
二、QEMU NFS 环境搭建

二、QEMU NFS 环境搭建

​ 在上一章节中，我们已经成功完成了内核和 busybox 环境的配置。为了进一步提高开发效率，我们可以使用 NFS（Network File System）来挂载根目录。NFS 允许我们将本地文件系统通过网络共享给虚拟机使用，这样在开发过程中…
阅读更多...
.NET 9.0 的 Blazor Web App 项目中 EF Core 【事务】使用备忘

.NET 9.0 的 Blazor Web App 项目中 EF Core 【事务】使用备忘

一、DbContext.Database.BeginTransactionAsync() 模式 1. 注意事项：连接字符串中启用了 MARS（Multiple Active Result Sets：MultipleActiveResultSetsTrue ）后，无法创建 保存点（保存点与 SQL Server 的多…
阅读更多...
记一次 Git Fetch 后切换分支为空的情况

记一次 Git Fetch 后切换分支为空的情况

Git Fetch 后切换分支为空的情况 在使用 Git 时，我遇到这样的情况：执行 git fetch 后切换分支，发现工作目录是空的，没有任何文件，所以插眼记录一下。 原因分析 git fetch 的作用：git fetch 只会从远程仓库…
阅读更多...
UMLS数据下载及访问

UMLS数据下载及访问

UMLS数据申请 这个直接在官网上申请即可，记得把地址填全，基本都会拿到lisence。 UMLS数据访问 UMLS的数据访问分为网页访问，API访问以及数据下载后的本地访问，网页访问，API访问按照官网的指示即可，这里主…
阅读更多...
使用 Docker 部署 Apache Spark 集群教程

使用 Docker 部署 Apache Spark 集群教程

简介 Apache Spark 是一个强大的统一分析引擎，用于大规模数据处理。本文将详细介绍如何使用 Docker 和 Docker Compose 快速部署一个包含一个 Master 节点和两个 Worker 节点的 Spark 集群。这种方法不仅简化了集群的搭建过程，还提供了资源隔离、易于扩…
阅读更多...
瑞萨RA-T系列芯片ADCGPT功能模块的配合使用

瑞萨RA-T系列芯片ADCGPT功能模块的配合使用

在马达或电源工程中，往往需要采集多路AD信号，且这些信号的优先级和采样时机不相同。本篇介绍在使用RA-T系列芯片建立马达或电源工程时，如何根据需求来设置主要功能模块ADC&GPT，包括采样通道打包和分组，GPT触发启动…
阅读更多...
20250217 随笔 redis非原子性操作简述

20250217 随笔 redis非原子性操作简述

从你提供的文本来看，核心是 Redis 作为缓存的检查机制，以及非原子性操作导致的不一致性问题。 我们可以拆解为两个部分来理解： 📌 1. 逻辑：先查 Redis，再决定是否注册 逻辑流程 先查询 Redis 是否有某个 …
阅读更多...
git-提交时间和作者时间的区别

git-提交时间和作者时间的区别

1.介绍 定义介绍 提交时间（Committer Date）：决定了提交在 Git 历史中的位置，通常影响 GitHub 上提交显示的顺序。 作者时间（Author Date）：虽然不影响提交的排序，但在每个提交详情页…
阅读更多...
PHP框架入门指南：从零构建现代Web应用

PHP框架入门指南：从零构建现代Web应用

一、为什么需要PHP框架？ 1.1 传统PHP开发的痛点 重复造轮子：用户认证、表单验证等基础功能需要反复开发代码混乱：缺乏统一结构导致维护困难安全漏洞：手动处理SQL注入/XSS攻击效率低下扩展性差：耦合代码难以适应业务增长1.2 框架的核心价值 标准化架构：MVC模式强制代码分…
阅读更多...
Leetcode 146 LRU缓存 的三种解法

Leetcode 146 LRU缓存 的三种解法

146. LRU 缓存 请你设计并实现一个满足 LRU (最近最少使用) 缓存 约束的数据结构。 实现 LRUCache 类： LRUCache(int capacity) 以 正整数 作为容量 capacity 初始化 LRU 缓存int get(int key) 如果关键字 key 存在于缓存中，则返回关键字的值&#xff0…
阅读更多...
尚硅谷 java 学习Day19 抽象类与抽象方法、接口、内部类

尚硅谷 java 学习Day19 抽象类与抽象方法、接口、内部类

6-5 抽象类(abstract)与抽象方法（important） 一、什么叫抽象类： 有时候将一个父类设计的非常抽象，以至于它没有具体的实例，这样的类称为抽象类 abstract关键字的使用： ​ 1、abstract:抽象的 ​ 2、abs…
阅读更多...
最新文章

Copyright @ 2022~2023