æ¦å¿µ
è¿æ¥æ± æ¯æ°æ®åºç¼ç¨ä¸å¸¸ç¨çä¸ç§ææ¯ï¼ç¨äºç®¡çæ°æ®åºè¿æ¥çå¤ç¨åéæ¾ï¼ä»èæé«æ°æ®åºæä½çæ§è½åæçãå¨Pythonä¸ï¼æ们å¯ä»¥ä½¿ç¨å¤ä¸ªç¬¬ä¸æ¹åºæ¥å®ç°è¿æ¥æ± çåè½ï¼ä¾å¦pymysqlãpsycopg2ãsqlalchemyçãæ¬æç¨å°ä»¥pymysqlåºä¸ºä¾ï¼ä»ç»å¦ä½å¨Pythonä¸ä½¿ç¨è¿æ¥æ± è¿è¡æ°æ®åºç¼ç¨ã
å®è·µæä½
æ¥éª¤1: å®è£ pymysqlåº
é¦å ï¼æ们éè¦å®è£ pymysqlåºï¼å¯ä»¥ä½¿ç¨pipå½ä»¤è¿è¡å®è£ ï¼å¦ä¸æ示ï¼
pip install pymysql
æ¥éª¤2: å¯¼å ¥pymysqlåºåè¿æ¥æ± 模å
å¨Python代ç ä¸ï¼æ们éè¦å¯¼å ¥pymysqlåºåpymysqlpool模åï¼åè æä¾äºè¿æ¥æ± çå®ç°ãå¯ä»¥ä½¿ç¨ä»¥ä¸æ¹å¼è¿è¡å¯¼å ¥ï¼
import pymysql
from pymysqlpool import ConnectionPool
æ¥éª¤3: å建è¿æ¥æ±
å¨ä½¿ç¨è¿æ¥æ± ä¹åï¼æ们éè¦å建ä¸ä¸ªè¿æ¥æ± 对象ï¼å¹¶è®¾ç½®è¿æ¥æ± çé ç½®åæ°ï¼ä¾å¦æ大è¿æ¥æ°ãæå°è¿æ¥æ°ãè¶ æ¶æ¶é´çã以ä¸æ¯ä¸ä¸ªå建è¿æ¥æ± ç示ä¾ï¼
# å建è¿æ¥æ± 对象
pool = ConnectionPool(
host='localhost', # æ°æ®åºä¸»æºå
port=3306, # æ°æ®åºç«¯å£å·
user='root', # æ°æ®åºç¨æ·å
password='password', # æ°æ®åºå¯ç
db='testdb', # æ°æ®åºå称
charset='utf8mb4', # æ°æ®åºå符é
max_connections=10, # æ大è¿æ¥æ°
min_connections=1, # æå°è¿æ¥æ°
max_idle_time=None, # è¿æ¥çæ大空é²æ¶é´ï¼åä½ç§
blocking=False, # å½è¿æ¥æ± 没æå¯ç¨è¿æ¥æ¶ï¼æ¯å¦é»å¡çå¾
autocommit=True, # è¿æ¥æ¯å¦èªå¨æ交äºå¡
ping=True, # æ¯æ¬¡ä»è¿æ¥æ± è·åè¿æ¥æ¶æ¯å¦èªå¨pingæ°æ®åº
cursorclass=pymysql.cursors.DictCursor # 游æ ç±»å
)
æ¥éª¤4: 使ç¨è¿æ¥æ± è¿è¡æ°æ®åºæä½
å¨å建è¿æ¥æ± åï¼æ们å¯ä»¥ä½¿ç¨è¿æ¥æ± 对象æ¥è·åæ°æ®åºè¿æ¥ï¼å¹¶æ§è¡æ°æ®åºæä½ãè¿æ¥æ± ä¼èªå¨ç®¡çè¿æ¥çå¤ç¨åéæ¾ï¼ä»èåå°äºæ¯æ¬¡è¿æ¥æ°æ®åºçå¼éã以ä¸æ¯ä¸ä¸ªä½¿ç¨è¿æ¥æ± è¿è¡æ°æ®åºæ¥è¯¢ç示ä¾ï¼
# ä»è¿æ¥æ± è·åè¿æ¥
conn = pool.get_conn()
cur = conn.cursor()
try:
# æ§è¡æ°æ®åºæ¥è¯¢æä½
cur.execute('SELECT * FROM users')
results = cur.fetchall()
for row in results:
print(row)
except Exception as e:
print("æ°æ®åºæ¥è¯¢åºéï¼", e)
finally:
# å
³é游æ åè¿æ¥
cur.close()
conn.close()
pool.release(conn) # å°è¿æ¥æ¾åè¿æ¥æ±
æ¥éª¤5: å ³éè¿æ¥æ±
å¨ç¨åºç»ææ¶ï¼æ们åºè¯¥å ³éè¿æ¥æ± ï¼ä»¥éæ¾è¿æ¥æ± å ç¨çèµæºãå¯ä»¥ä½¿ç¨ä»¥ä¸æ¹å¼æ¥å ³éè¿æ¥æ± ï¼
pool.close()
å®æ´çPythonè¿æ¥æ± 示ä¾ä»£ç å¦ä¸ï¼
import pymysql
from pymysqlpool import ConnectionPool
# å建è¿æ¥æ± 对象
pool = ConnectionPool(
host='localhost', # æ°æ®åºä¸»æºå
port=3306, # æ°æ®åºç«¯å£å·
user='root', # æ°æ®åºç¨æ·å
password='password', # æ°æ®åºå¯ç
db='testdb', # æ°æ®åºå称
charset='utf8mb4', # æ°æ®åºå符é
max_connections=10, # æ大è¿æ¥æ°
min_connections=1, # æå°è¿æ¥æ°
max_idle_time=None, # è¿æ¥çæ大空é²æ¶é´ï¼åä½ç§
blocking=False, # å½è¿æ¥æ± 没æå¯ç¨è¿æ¥æ¶ï¼æ¯å¦é»å¡çå¾
autocommit=True, # è¿æ¥æ¯å¦èªå¨æ交äºå¡
ping=True, # æ¯æ¬¡ä»è¿æ¥æ± è·åè¿æ¥æ¶æ¯å¦èªå¨pingæ°æ®åº
cursorclass=pymysql.cursors.DictCursor # 游æ ç±»å
)
# ä»è¿æ¥æ± è·åè¿æ¥
conn = pool.get_conn()
cur = conn.cursor()
try:
# æ§è¡æ°æ®åºæ¥è¯¢æä½
cur.execute('SELECT * FROM users')
results = cur.fetchall()
for row in results:
print(row)
except Exception as e:
print("æ°æ®åºæ¥è¯¢åºéï¼", e)
finally:
# å
³é游æ åè¿æ¥
cur.close()
conn.close()
pool.release(conn) # å°è¿æ¥æ¾åè¿æ¥æ±
# å
³éè¿æ¥æ±
pool.close()
æ»ç»
éè¿ä½¿ç¨è¿æ¥æ± ï¼æ们å¯ä»¥å¨Pythonæ°æ®åºç¼ç¨ä¸å®ç°è¿æ¥çå¤ç¨åéæ¾ï¼ä»èæé«æ°æ®åºæä½çæ§è½åæçãåæ¶ï¼è¿æ¥æ± è¿æä¾äºè®¸å¤é ç½®åæ°ï¼å¯ä»¥æ ¹æ®å®é éæ±è¿è¡è°æ´ï¼ä»¥æ»¡è¶³ä¸ååºæ¯ä¸çæ°æ®åºè¿æ¥ç®¡çéæ±ãå¸æè¿ä¸ªæç¨å¯ä»¥å¸®å©æ¨ææ¡å¨Pythonä¸ä½¿ç¨è¿æ¥æ± è¿è¡æ°æ®åºç¼ç¨çåºæ¬æ¥éª¤å注æäºé¡¹ã