求数据库连接池推荐

问答 / 0 / 4 / 创建于 2年前

需要一个类似SMProxy，无需改动代码，只需要简单配置就能使用的mysql数据库连接池

不论是扩展还是插件，又或是其他语言的程序都可以。

iGuoji

0 声望

暂无个人描述~

0 人点赞

你将学到如 RESTFul 设计风格、PostMan 的使用、OAuth 流程，JWT 概念及使用和 API 开发相关的进阶知识。

以构建论坛项目 LaraBBS 为线索，展开对 Laravel 框架的全面学习。应用程序架构思路贴近 Laravel 框架的设计哲学。

讨论数量: 4

微加加的朋友

课程读者 114 声望

以下是一个基于Python的MySQL数据库连接池的示例，可以类比实现类似SMProxy的功能。具体实现方式可以根据实际需要进行调整和改进。

import pymysql
from queue import Queue
from threading import Lock

class MySQLPool:
    def __init__(self, max_conn, host, port, user, password, database):
        self._max_conn = max_conn  # 最大连接数
        self._host = host
        self._port = port
        self._user = user
        self._password = password
        self._database = database
        self._pool = Queue(max_conn)  # 连接池
        self._lock = Lock()  # 连接池锁

        # 创建初始连接
        for _ in range(max_conn):
            conn = self._create_connection()
            if conn:
                self._pool.put(conn)

    # 获取数据库连接
    def get_connection(self):
        conn = self._pool.get()
        if conn:
            return conn
        else:
            return self._create_connection()

    # 释放数据库连接
    def release_connection(self, conn):
        if conn:
            self._pool.put(conn)

    # 创建数据库连接
    def _create_connection(self):
        try:
            conn = pymysql.connect(host=self._host, port=self._port,
                                   user=self._user, password=self._password,
                                   database=self._database)
            return conn
        except Exception as e:
            print(e)
            return None

    # 执行数据库查询
    def query(self, sql, params=None):
        conn = self.get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute(sql, params)
            result = cursor.fetchall()
            cursor.close()
            return result
        except Exception as e:
            print(e)
            return None
        finally:
            self.release_connection(conn)

    # 执行数据库修改
    def execute(self, sql, params=None):
        conn = self.get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute(sql, params)
            conn.commit()
            cursor.close()
            return True
        except Exception as e:
            print(e)
            return False
        finally:
            self.release_connection(conn)

使用时可以简单地实例化MySQLPool类，并调用其中的方法来进行数据库操作，如下所示：

pool = MySQLPool(max_conn=10, host='localhost', port=3306, user='root', password='password', database='test')

# 查询数据
result = pool.query('SELECT * FROM `users` WHERE `id`=%s', (1,))
print(result)

# 修改数据
result = pool.execute('UPDATE `users` SET `name`=%s WHERE `id`=%s', ('Tom', 1))
print(result)