python連接池pooledDB源碼閱讀參數(shù)的使用

更新時(shí)間：2024年07月18日 09:29:28 作者：wenweny2020

這篇文章主要介紹了python連接池pooledDB源碼閱讀參數(shù)的使用,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

pooledDB參數(shù)詳解

from DBUtils.PooledDB import PooledDB

self.__pool = PooledDB(creator=pymysql,
                       mincached=1, 
                       maxcached=4, # 連接池中最大空閑連接數(shù)
                       maxconnections=4,#允許的最大連接數(shù)
                       blocking=True,# 設(shè)置為true，則阻塞并等待直到連接數(shù)量減少，false默認(rèn)情況下將報(bào)告錯(cuò)誤。
                       ping=1,#默認(rèn)=1表示每當(dāng)從池中獲取時(shí)，使用ping()檢查連接
                       host=self.host,
                       port=self.port,
                       user=self.user,
                       passwd=self.passwd,
                       db=self.db_name,
                       charset=self.charset
                      )

mincached：連接池中的初始空閑連接，默認(rèn)0或None表示創(chuàng)建連接池時(shí)沒有連接。但對(duì)照源碼及實(shí)驗(yàn)效果來看，這個(gè)參數(shù)并沒有起作用。

# PooledDB.py源碼 267行
idle = [self.dedicated_connection() for i in range(mincached)]
while idle:
    idle.pop().close()
# 確實(shí)是創(chuàng)建連接池時(shí)創(chuàng)建了mincached個(gè)連接，但返回之前都關(guān)閉了。所以創(chuàng)建好的時(shí)候并沒有mincached個(gè)初始連接

maxcached：連接池中最大空閑連接數(shù)，默認(rèn)0或None表示沒有連接池大小限制
maxshared：最大共享連接數(shù)。默認(rèn)0或None表示所有連接都是專用的
maxconnections：最大允許連接數(shù)，默認(rèn)0或None表示沒有連接限制

# PooledDB.py源碼 255行
if maxconnections:
	if maxconnections < maxcached:
		maxconnections = maxcached
	if maxconnections < maxshared:
		maxconnections = maxshared
	self._maxconnections = maxconnections
else:
	self._maxconnections = 0
# maxcached、maxshared同時(shí)影響maxconnections
# maxconnections=max(maxcached, maxshared)

# PooledDB.py源碼 356行
	# 當(dāng)收到一個(gè)連接放回請(qǐng)求時(shí)
    # if 沒有最大空閑連接數(shù)限制，或現(xiàn)在的空閑連接數(shù)小于最大空閑連接數(shù)，則將事務(wù)回滾，并將這個(gè)連接放回空閑連接處；
    # else:直接關(guān)閉
    def cache(self, con):
        """Put a dedicated專用 connection back into the idle空閑 cache."""
        self._lock.acquire()
        try:
            if not self._maxcached or len(self._idle_cache) < self._maxcached:
                con._reset(force=self._reset)  # rollback possible transaction
                # the idle cache is not full, so put it there
                self._idle_cache.append(con)  # append it to the idle cache
            else:  # if the idle cache is already full,
                con.close()  # then close the connection
            self._connections -= 1
            self._lock.notify()
        finally:
            self._lock.release()
# cache方法被使用
    def close(self):
        """Close the pooled dedicated connection."""
        # Instead of actually closing the connection,
        # return it to the pool for future reuse.
        if self._con:
            self._pool.cache(self._con)
            self._con = None

blocking：True表示沒有空閑可用連接時(shí)，堵塞并等待；False表示直接報(bào)錯(cuò)。默認(rèn)為False。
maximum：?jiǎn)蝹€(gè)連接的最大reuse次數(shù)，默認(rèn)0或None表示無限重復(fù)使用，當(dāng)達(dá)到連接大最大使用次數(shù)，連接將被重置。

# SteadyDB.py 483行
if self._maxusage:
	if self._usage >= self._maxusage:
        # the connection was used too often
        raise self._failure
cursor = self._con.cursor(*args, **kwargs)  # try to get a cursor

setsession: optional list of SQL commands that may serve to prepare the session, 在連接的時(shí)候就會(huì)被執(zhí)行的sql語句。

# SteadyDB.py 298行
    def _setsession(self, con=None):
        """Execute the SQL commands for session preparation."""
        if con is None:
            con = self._con
        if self._setsession_sql:
            cursor = con.cursor()
            for sql in self._setsession_sql:
                cursor.execute(sql)
            cursor.close()

reset：連接放回連接池中時(shí)是如何被重置的，默認(rèn)為True。self._transaction僅在begin()內(nèi)被置為True。默認(rèn)為True時(shí)，true的話每次返回連接池都會(huì)回滾事務(wù)，F(xiàn)alse的話只會(huì)回滾begin()顯式開啟的事務(wù).

    def cache(self, con):
        """Put a dedicated connection back into the idle cache."""
        self._lock.acquire()
        try:
            if not self._maxcached or len(self._idle_cache) < self._maxcached:
                con._reset(force=self._reset)  # rollback possible transaction
                # the idle cache is not full, so put it there
                self._idle_cache.append(con)  # append it to the idle cache
            else:  # if the idle cache is already full,
                con.close()  # then close the connection
            self._connections -= 1
            self._lock.notify()
        finally:
            self._lock.release()
 
    def _reset(self, force=False):
        """Reset a tough connection.

        Rollback if forced or the connection was in a transaction.

        """
        if not self._closed and (force or self._transaction):
            try:
                self.rollback()
            except Exception:
                pass
            
    def begin(self, *args, **kwargs):
        """Indicate the beginning of a transaction.

        During a transaction, connections won't be transparently
        replaced, and all errors will be raised to the application.

        If the underlying driver supports this method, it will be called
        with the given parameters (e.g. for distributed transactions).

        """
        self._transaction = True
        try:
            begin = self._con.begin
        except AttributeError:
            pass
        else:
            begin(*args, **kwargs)

failures：異常類補(bǔ)充，如果(OperationalError, InternalError)這兩個(gè)不夠。

except self._failures as error:

ping: 官方解釋是 (0 = None = never, 1 = default = when _ping_check() is called, 2 = whenever a cursor is created, 4 = when a query is executed, 7 = always, and all other bit combinations of these values 是上面情況的集合)，但在源碼中只區(qū)分了是否非零，似乎數(shù)值多少?zèng)]有太大意義。

    def _ping_check(self, ping=1, reconnect=True):
        """Check whether the connection is still alive using ping().

        If the the underlying connection is not active and the ping
        parameter is set accordingly, the connection will be recreated
        unless the connection is currently inside a transaction.

        """
        if ping & self._ping:
            try:  # if possible, ping the connection
                alive = self._con.ping()
            except (AttributeError, IndexError, TypeError, ValueError):
                self._ping = 0  # ping() is not available
                alive = None
                reconnect = False
            except Exception:
                alive = False
            else:
                if alive is None:
                    alive = True
                if alive:
                    reconnect = False
            if reconnect and not self._transaction:
                try:  # try to reopen the connection
                    con = self._create()
                except Exception:
                    pass
                else:
                    self._close()
                    self._store(con)
                    alive = True
            return alive

使用方法

    def start_conn(self):
        try:
            # maxshared 允許的最大共享連接數(shù)，默認(rèn)0/None表示所有連接都是專用的
            # 當(dāng)線程關(guān)閉不再共享的連接時(shí)，它將返回到空閑連接池中，以便可以再次對(duì)其進(jìn)行回收。
            # mincached 連接池中空閑連接的初始連接數(shù)，實(shí)驗(yàn)證明沒啥用
            self.__pool = PooledDB(creator=pymysql,
                                   mincached=1, # mincached 連接池中空閑連接的初始連接數(shù)，但其實(shí)沒用
                                   maxcached=4,  # 連接池中最大空閑連接數(shù)
                                   maxshared=3, #允許的最大共享連接數(shù)
                                   maxconnections=2,  # 允許的最大連接數(shù)
                                   blocking=False,  # 設(shè)置為true，則阻塞并等待直到連接數(shù)量減少，false默認(rèn)情況下將報(bào)告錯(cuò)誤。
                                   host=self.host,
                                   port=self.port,
                                   user=self.user,
                                   passwd=self.passwd,
                                   db=self.db_name,
                                   charset=self.charset
                                   )
            print("0 start_conn連接數(shù):%s " % (self.__pool._connections))
            self.conn = self.__pool.connection()
            print('connect success')
            print("1 start_conn連接數(shù):%s " % (self.__pool._connections))

            self.conn2 = self.__pool.connection()
            print("2 start_conn連接數(shù):%s " % (self.__pool._connections))
            db3 = self.__pool.connection()
            print("3 start_conn連接數(shù):%s " % (self.__pool._connections))
            db4 = self.__pool.connection()
            print("4 start_conn連接數(shù):%s " % (self.__pool._connections))
            db5 = self.__pool.connection()
            print("5 start_conn連接數(shù):%s " % (self.__pool._connections))
            # self.conn.close()
            print("6 start_conn連接數(shù):%s " % (self.__pool._connections))
            return True
        except:
            print('connect failed')
            return False

0 start_conn連接數(shù):0
connect success
1 start_conn連接數(shù):1
2 start_conn連接數(shù):2
3 start_conn連接數(shù):3
4 start_conn連接數(shù):4
connect failed

如上程序，可對(duì)照試驗(yàn)結(jié)果，詳細(xì)理解一下上述的幾個(gè)參數(shù)。

mincached確實(shí)沒用，pooledDB對(duì)象生成退出后，并沒有mincached個(gè)初始化連接。
maxconnections = max(maxcached,maxshared)，對(duì)照結(jié)果來看，最大的連接數(shù)顯然等于maxcached,maxshared的較大者4，所以可以連續(xù)開四個(gè)連接，但到第5個(gè)時(shí)顯示連接失敗。
若將blocking改為True，則實(shí)驗(yàn)結(jié)果最后一行的”connect failed“不會(huì)出現(xiàn)，程序會(huì)一直堵塞等待新的空閑連接出現(xiàn)，在本例中，沒有操作關(guān)閉原有連接，程序會(huì)一直堵塞等待。

參考資料：

DBUtils官網(wǎng)資料