1.初始化

def __init__(self, **kwargs):

self.size = kwargs.get('size', 10)

self.kwargs = kwargs

self.conn_queue = queue.Queue(maxsize=self.size)

for i in range(self.size):

self.conn_queue.put(self._create_new_conn())

size：連接池支持的連接數，這里定義為10

conn_queue：定義了一個隊列，隊列存放的是數據庫的連接

for循環：建立好十個與數據庫的連接，把這些連接放到隊列中

2.私有方法:連接數據庫

def _create_new_conn(self):

return MySQLdb.connect(host=self.kwargs.get('host', '127.0.0.1'),

user=self.kwargs.get('user'),

passwd=self.kwargs.get('password'),

port=self.kwargs.get('port', 3306),

connect_timeout=5)

connect_timeout：在獲取連接階段起作用

獲取MySQL連接是多次握手的結果，除了用戶名和密碼的匹配校驗外，還有IP->HOST->DNS->IP驗證，任何一步都可能因為網絡問題導致線程阻塞。為了防止線程浪費在不必要的校驗等待上，超過connect_timeout的連接請求將會被拒絕。

官方描述：connect_timeout(The number of seconds that the mysqld server waits for a connect packet before responding with Bad handshake. The default value is 10 seconds)

這里設置了5秒。

3.私有方法：put

def _put_conn(self, conn):

self.conn_queue.put(conn)

使用了queue的put方法，將連接放到隊列中，put默認是阻塞調用，非阻塞版本為put_nowait()方法，相當于put(conn, False)

阻塞調用是指：調用返回結果之前，當前線程會被掛起，線程進入非可執行狀態，在這個狀態下CPU不會給線程分配時間片，線程暫停運行。函數只有在得到結果之后才會返回。

4.私有方法：獲取連接

def _get_conn(self):

conn = self.conn_queue.get()

if conn is None:

self._create_new_conn()

return conn

當要連接數據庫時，不再創建新的連接，而是之間從隊列中獲取連接，用完之后再把連接放回隊列中，如果獲取的連接為空，再新建連接。

get():如果隊列為空，get會等待，直到隊列里有數據以后再取值，get取值會在隊列中移除一個數據，所以當取完連接用完之后，要再使用put方法把連接放回連接池。（get默認為阻塞調用，非阻塞調用方法為get_nowait()）

get_nowait()：取值的時候不等待，如果取不到值，程序直接崩潰，所以在獲取隊列的數據的時候要統一使用get，代碼才不會有問題

使用get_nowait()和put_nowait()的時候要做捕獲異常處理。

5.執行SQL語句函數

def exec_sql(self, sql):

conn = self._get_conn()

try:

with conn as cur:

cur.execute(sql)

return cur.fetchall()

except MySQLdb.ProgrammingError as e:

LOG.error("execute sql ({0}) error {1}".format(sql, e))

raise e

except MySQLdb.OperationalError as e:

conn = self._create_new_conn()

raise e

finally:

self._put_conn(conn)

傳遞進來的參數是SQL語句

獲取連接后，一般要創建游標，然后使用游標再進行對數據的增刪改查，這里使用了with語句，就不用再創建游標了。cur執行SQL語句，然后使用fetchall返回所有匹配的每個元素，每個元素作為一個元組組成一個大元組，最后返回的是這一個大元組。

然后捕獲了兩個異常，一個是語法錯誤異常，另一個是編碼問題的異常。

最后無論是正常建立連接執行了SQL語句，還是發生了異常，都要把連接放回到連接隊列中去。

6.刪除連接

def __del__(self):

try:

while True:

conn = self.conn_queue.get_nowait()

if conn:

conn.close()

except queue.Empty:

pass

獲取到連接之后用close關閉連接，如果取到的隊列中的連接已經為空了，直接pass

其中queue.Empty的作用是，如果隊列為空，返回True，如果不為空，返回False。

7.完整代碼

# -*- coding:UTF-8 -*-

import queue

import MySQLdb

class ConnectionPool(object):

def __init__(self, **kwargs):

self.size = kwargs.get('size', 10)

self.kwargs = kwargs

self.conn_queue = queue.Queue(maxsize=self.size)

for i in range(self.size):

self.conn_queue.put(self._create_new_conn())

def _create_new_conn(self):

return MySQLdb.connect(host=self.kwargs.get('host', '127.0.0.1'),

user=self.kwargs.get('user'),

passwd=self.kwargs.get('password'),

port=self.kwargs.get('port', 3306),

connect_timeout=5)

def _put_conn(self, conn):

self.conn_queue.put(conn)

def _get_conn(self):

conn = self.conn_queue.get()

if conn is None:

self._create_new_conn()

return conn

def exec_sql(self, sql):

conn = self._get_conn()

try:

with conn as cur:

cur.execute(sql)

return cur.fetchall()

except MySQLdb.ProgrammingError as e:

# 可以加一行將異常記錄到日志中

raise e

except MySQLdb.OperationalError as e:

conn = self._create_new_conn()

raise e

finally:

self._put_conn(conn)

def __del__(self):

try:

while True:

conn = self.conn_queue.get_nowait()

if conn:

conn.close()

except queue.Empty:

pass

總結

以上是生活随笔為你收集整理的python数据库连接池_Python实现数据库连接池的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。