python 數據庫連接池

對於一個簡單的數據庫應用，由於對於數據庫的訪問不是很頻繁。這時可以簡單地在需要訪問數據庫時，就新創建一個連接，用完后就關閉它，這樣做也不會帶來什么明顯的性能上的開銷。但是對於一個復雜的數據庫應用，情況就完全不同了。頻繁的建立、關閉連接，會極大的減低系統的性能，因為對於連接的使用成了系統性能的瓶頸。

連接復用。通過建立一個數據庫連接池以及一套連接使用管理策略，使得一個數據庫連接可以得到高效、安全的復用，避免了數據庫連接頻繁建立、關閉的開銷。

對於共享資源，有一個很著名的設計模式：資源池。該模式正是為了解決資源頻繁分配、釋放所造成的問題的。把該模式應用到數據庫連接管理領域，就是建立一個數據庫連接池，提供一套高效的連接分配、使用策略，最終目標是實現連接的高效、安全的復用。


數據庫連接池的基本原理是在內部對象池中維護一定數量的數據庫連接，並對外暴露數據庫連接獲取和返回方法。如：

外部使用者可通過getConnection 方法獲取連接，使用完畢后再通過releaseConnection 方法將連接返回，注意此時連接並沒有關閉，而是由連接池管理器回收，並為下一次使用做好准備。

 

數據庫連接池技術帶來的優勢：

1． 資源重用

由於數據庫連接得到重用，避免了頻繁創建、釋放連接引起的大量性能開銷。在減少系統消耗的基礎上，另一方面也增進了系統運行環境的平穩性（減少內存碎片以及數據庫臨時進程/線程的數量）。

2． 更快的系統響應速度

數據庫連接池在初始化過程中，往往已經創建了若干數據庫連接置於池中備用。此時連接的初始化工作均已完成。對於業務請求處理而言，直接利用現有可用連接，避免了數據庫連接初始化和釋放過程的時間開銷，從而縮減了系統整體響應時間。

3． 新的資源分配手段

對於多應用共享同一數據庫的系統而言，可在應用層通過數據庫連接的配置，實現數據庫連接池技術，幾年錢也許還是個新鮮話題，對於目前的業務系統而言，如果設計中還沒有考慮到連接池的應用，那么…….快在設計文檔中加上這部分的內容吧。某一應用最大可用數據庫連接數的限制，避免某一應用獨占所有數據庫資源。

4． 統一的連接管理，避免數據庫連接泄漏

在較為完備的數據庫連接池實現中，可根據預先的連接占用超時設定，強制收回被占用連接。從而避免了常規數據庫連接操作中可能出現的資源泄漏。一個最小化的數據庫連接池實現：

 

import pymysql
import os
import configparser
from pymysql.cursors import DictCursor
from dbutils.pooled_db import PooledDB


class Config(object):
    """
    # Config().get_content("user_information")
    配置文件里面的參數
    [dbMysql]
    host = 192.168.1.180
    port = 3306
    user = root
    password = 123456
    """

    def __init__(self, config_filename="config.ini"):
        file_path = os.path.join(os.path.dirname(__file__), config_filename)
        self.cf = configparser.ConfigParser()
        self.cf.read(file_path)

    def get_sections(self):
        return self.cf.sections()

    def get_options(self, section):
        return self.cf.options(section)

    def get_content(self, section):
        result = {}
        for option in self.get_options(section):
            value = self.cf.get(section, option)
            result[option] = int(value) if value.isdigit() else value
        return result


class BasePymysqlPool(object):
    def __init__(self, host, port, user, password, db_name):
        self.db_host = host
        self.db_port = int(port)
        self.user = user
        self.password = str(password)
        self.db = db_name
        self.conn = None
        self.cursor = None


class MyPymysqlPool(BasePymysqlPool):
    """
    MYSQL數據庫對象，負責產生數據庫連接 , 此類中的連接采用連接池實現
        獲取連接對象：conn = Mysql.getConn()
        釋放連接對象;conn.close()或del conn
    """
    # 連接池對象
    __pool = None

    def __init__(self, conf_name=None):
        self.conf = Config().get_content(conf_name)
        super(MyPymysqlPool, self).__init__(**self.conf)
        # 數據庫構造函數，從連接池中取出連接，並生成操作游標
        self._conn = self.__getConn()
        self._cursor = self._conn.cursor()

    def __getConn(self):
        """
        @summary: 靜態方法，從連接池中取出連接
        @return MySQLdb.connection
        """
        if MyPymysqlPool.__pool is None:
            __pool = PooledDB(creator=pymysql,
                              mincached=1,
                              maxcached=20,
                              host=self.db_host,
                              port=self.db_port,
                              user=self.user,
                              passwd=self.password,
                              db=self.db,
                              use_unicode=True,
                              charset="utf8",
                              cursorclass=DictCursor)
            # print("12211212")
        return __pool.connection()

    def getAll(self, sql, param=None):
        """
        @summary: 執行查詢，並取出所有結果集
        @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，並將條件值使用參數[param]傳遞進來
        @param param: 可選參數，條件列表值（元組/列表）
        @return: result list(字典對象)/boolean 查詢到的結果集
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchall()
        else:
            result = False
        return result

    def getOne(self, sql, param=None):
        """
        @summary: 執行查詢，並取出第一條
        @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，並將條件值使用參數[param]傳遞進來
        @param param: 可選參數，條件列表值（元組/列表）
        @return: result list/boolean 查詢到的結果集
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchone()
        else:
            result = False
        return result

    def getMany(self, sql, num, param=None):
        """
        @summary: 執行查詢，並取出num條結果
        @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，並將條件值使用參數[param]傳遞進來
        @param num:取得的結果條數
        @param param: 可選參數，條件列表值（元組/列表）
        @return: result list/boolean 查詢到的結果集
        """
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        if count > 0:
            result = self._cursor.fetchmany(num)
        else:
            result = False
        return result

    def insertMany(self, sql, values):
        """
        @summary: 向數據表插入多條記錄
        @param sql:要插入的ＳＱＬ格式
        @param values:要插入的記錄數據tuple(tuple)/list[list]
        @return: count 受影響的行數
        """
        count = self._cursor.executemany(sql, values)
        return count

    def __query(self, sql, param=None):
        if param is None:
            count = self._cursor.execute(sql)
        else:
            count = self._cursor.execute(sql, param)
        return count

    def update(self, sql, param=None):
        """
        @summary: 更新數據表記錄
        @param sql: ＳＱＬ格式及條件，使用(%s,%s)
        @param param: 要更新的  值 tuple/list
        @return: count 受影響的行數
        """
        return self.__query(sql, param)

    def insert(self, sql, param=None):
        """
        @summary: 更新數據表記錄
        @param sql: ＳＱＬ格式及條件，使用(%s,%s)
        @param param: 要更新的  值 tuple/list
        @return: count 受影響的行數
        """
        return self.__query(sql, param)

    def delete(self, sql, param=None):
        """
        @summary: 刪除數據表記錄
        @param sql: ＳＱＬ格式及條件，使用(%s,%s)
        @param param: 要刪除的條件 值 tuple/list
        @return: count 受影響的行數
        """
        return self.__query(sql, param)

    def begin(self):
        """
        @summary: 開啟事務
        """
        self._conn.autocommit(0)

    def end(self, option='commit'):
        """
        @summary: 結束事務
        """
        if option == 'commit':
            self._conn.commit()
        else:
            self._conn.rollback()

    def dispose(self, isEnd=1):
        """
        @summary: 釋放連接池資源
        """
        if False not in isEnd:
            self.end('commit')
        else:
            print("sql錯誤，rollback")
            self.end('rollback')
        self._cursor.close()
        self._conn.close()


# 調用方法
if __name__ == '__main__':
    mysql = MyPymysqlPool("dbMysql")
    sqlAll = "show tables;"
    result = mysql.getAll(sqlAll)
    print(result)
    # 釋放資源
    mysql.dispose(isEnd=[result, result])

配置mysql參數

[dbMysql]
host = 192.168.1.2
port = 3306
user = root
password = root
db_name = test