python连接池pooledDB源码阅读参数的使用

1

2

3

4

5

6

7

8

9

10

11

12

13

self.__pool = PooledDB(creator=pymysql,

                       mincached=1,

                       maxcached=4, # 连接池中最大空闲连接数

                       maxconnections=4,#允许的最大连接数

                       blocking=True,# 设置为true，则阻塞并等待直到连接数量减少，false默认情况下将报告错误。

                       ping=1,#默认=1表示每当从池中获取时，使用ping()检查连接

                       host=self.host,

                       port=self.port,

                       user=self.user,

                       passwd=self.passwd,

                       db=self.db_name,

                       charset=self.charset

                      )

1

2

3

4

5

# PooledDB.py源码 267行

idle = [self.dedicated_connection() for i in range(mincached)]

while idle:

    idle.pop().close()

# 确实是创建连接池时创建了mincached个连接，但返回之前都关闭了。所以创建好的时候并没有mincached个初始连接

1

2

3

4

5

6

7

8

9

10

11

# PooledDB.py源码 255行

if maxconnections:

    if maxconnections < maxcached:

        maxconnections = maxcached

    if maxconnections < maxshared:

        maxconnections = maxshared

    self._maxconnections = maxconnections

else:

    self._maxconnections = 0

# maxcached、maxshared同时影响maxconnections

# maxconnections=max(maxcached, maxshared)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

# PooledDB.py源码 356行

    # 当收到一个连接放回请求时

    # if 没有最大空闲连接数限制，或现在的空闲连接数小于最大空闲连接数，则将事务回滚，并将这个连接放回空闲连接处；

    # else:直接关闭

    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

1

2

3

4

5

6

# 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

1

2

3

4

5

6

7

8

9

10

# 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()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

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)

1

except self._failures as error:

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

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

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

def start_conn(self):

    try:

        # maxshared 允许的最大共享连接数，默认0/None表示所有连接都是专用的

        # 当线程关闭不再共享的连接时，它将返回到空闲连接池中，以便可以再次对其进行回收。

        # mincached 连接池中空闲连接的初始连接数，实验证明没啥用

        self.__pool = PooledDB(creator=pymysql,

                               mincached=1, # mincached 连接池中空闲连接的初始连接数，但其实没用

                               maxcached=4,  # 连接池中最大空闲连接数

                               maxshared=3, #允许的最大共享连接数

                               maxconnections=2,  # 允许的最大连接数

                               blocking=False,  # 设置为true，则阻塞并等待直到连接数量减少，false默认情况下将报告错误。

                               host=self.host,

                               port=self.port,

                               user=self.user,

                               passwd=self.passwd,

                               db=self.db_name,

                               charset=self.charset

                               )

        print("0 start_conn连接数:%s " % (self.__pool._connections))

        self.conn = self.__pool.connection()

        print('connect success')

        print("1 start_conn连接数:%s " % (self.__pool._connections))

        self.conn2 = self.__pool.connection()

        print("2 start_conn连接数:%s " % (self.__pool._connections))

        db3 = self.__pool.connection()

        print("3 start_conn连接数:%s " % (self.__pool._connections))

        db4 = self.__pool.connection()

        print("4 start_conn连接数:%s " % (self.__pool._connections))

        db5 = self.__pool.connection()

        print("5 start_conn连接数:%s " % (self.__pool._connections))

        # self.conn.close()

        print("6 start_conn连接数:%s " % (self.__pool._connections))

        return True

    except:

        print('connect failed')

        return False