開篇：為什么需要這篇指南

去年做個(gè)數(shù)據(jù)分析平臺，后端用Python開發(fā)，要連金倉數(shù)據(jù)庫做數(shù)據(jù)清洗和統(tǒng)計(jì)。一開始圖省事，用了普通數(shù)據(jù)庫驅(qū)動(dòng)，結(jié)果遇到一堆麻煩：

• 代碼到處都是try-catch-finally，看著都煩
• 并發(fā)一上來，連接動(dòng)不動(dòng)就超時(shí)
• 最要命的是數(shù)據(jù)庫主備切換時(shí)，應(yīng)用直接掛了

后來換成了ksycopg2驅(qū)動(dòng)，這些問題基本都解決了。這篇分上下兩篇，上篇說說連接管理、高可用配置和數(shù)據(jù)類型映射，下篇講SQL執(zhí)行、批量操作和擴(kuò)展功能。

一、ksycopg2 是什么

ksycopg2是金倉數(shù)據(jù)庫的Python驅(qū)動(dòng)，用C語言寫的，性能不錯(cuò)，完整實(shí)現(xiàn)了Python DB API 2.0規(guī)范。說白了，就是Python連接金倉數(shù)據(jù)庫的標(biāo)準(zhǔn)方式。

它依賴libkci庫，libkci又依賴OpenSSL。所以安裝時(shí)要是報(bào)缺庫，多半是這兩個(gè)依賴沒配好。

二、安裝配置

2.1 版本選擇

ksycopg2的版本號得和Python版本對上。比如包名是ksycopg2?python2.7，就只能用在Python 2.7上。

官方支持的情況：

我自己用的是Python 3.9，對應(yīng)的驅(qū)動(dòng)跑得挺穩(wěn)。

2.2 安裝步驟

最簡單的就是用pip裝。不過得先確認(rèn)pip對應(yīng)哪個(gè)Python版本。

# 查看 pip 對應(yīng)的 Python 版本
pip --version
# 安裝 ksycopg2
pip install ksycopg2

要是pip裝不了，或者要離線安裝，就去金倉官網(wǎng)下載對應(yīng)版本的驅(qū)動(dòng)包，解壓后把ksycopg2文件夾扔到Python的site-packages目錄下。

怎么找這個(gè)目錄？在Python環(huán)境里執(zhí)行：

import sys
sys.path

輸出的列表里，帶site-packages的那個(gè)路徑就是了。

2.3 依賴庫配置

ksycopg2依賴libkci.so（Linux）或libkci.dll（Windows）。驅(qū)動(dòng)裝好了但導(dǎo)入報(bào)錯(cuò)，通常是沒找到這些庫。

報(bào)錯(cuò)信息長這樣：

ImportError: libkci.so: cannot open shared object file: No such file or directory

解決辦法：把libkci所在目錄加到LD_LIBRARY_PATH環(huán)境變量里。

export LD_LIBRARY_PATH=/home/kingbase/lib:$LD_LIBRARY_PATH

Windows上就把libkci.dll放到Python安裝目錄或系統(tǒng)PATH里。

一個(gè)常見坑：Windows下導(dǎo)入ksycopg2時(shí)報(bào)DLL load failed，多半是缺MSVC運(yùn)行庫。裝上Visual C++ Redistributable就好了。

2.4 驗(yàn)證安裝

import ksycopg2
print(ksycopg2.__version__)

不報(bào)錯(cuò)且輸出版本號，說明裝好了。

三、連接數(shù)據(jù)庫

3.1 三種連接方式

方式一：鍵值對字符串

import ksycopg2

conn = ksycopg2.connect(
    "dbname=TEST user=SYSTEM password=123456 host=127.0.0.1 port=54321"
)

方式二：關(guān)鍵字參數(shù)

這種方式代碼更清晰，生產(chǎn)環(huán)境推薦用：

conn = ksycopg2.connect(
    database='TEST',
    user='SYSTEM',
    password='123456',
    host='127.0.0.1',
    port='54321',
    client_encoding='UTF8'      # 設(shè)置客戶端編碼
)

方式三：URI 格式

conn = ksycopg2.connect(
    "kingbase://SYSTEM:123456@127.0.0.1:54321/TEST"
)

還有一種URI格式：

conn = ksycopg2.connect(
    "postgresql://SYSTEM:123456@127.0.0.1:54321/TEST"
)

支持postgresql前綴主要是歷史原因，實(shí)際連的還是金倉。

3.2 連接參數(shù)說明

其中connect_timeout建議設(shè)一下，默認(rèn)無限等待在某些網(wǎng)絡(luò)環(huán)境下會(huì)卡住。

3.3 關(guān)閉連接

用完記得關(guān)，不然會(huì)占著數(shù)據(jù)庫連接數(shù)：

conn.close()

可以用with語句自動(dòng)管理資源：

with ksycopg2.connect(database='TEST', user='SYSTEM') as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT 1")
        print(cur.fetchone())
# 退出 with 塊時(shí)自動(dòng)提交事務(wù)，但連接不會(huì)自動(dòng)關(guān)閉，還得手動(dòng) conn.close()

四、高可用配置（重點(diǎn)）

這是ksycopg2比較實(shí)用的功能之一。如果你的金倉數(shù)據(jù)庫是集群部署（主備模式），配好高可用參數(shù)，主備切換時(shí)應(yīng)用就能自動(dòng)重連。

4.1 多主機(jī)配置

先看基礎(chǔ)的多主機(jī)配置：

# 配置兩個(gè)主機(jī)地址
hosts = "192.168.1.100,192.168.1.101"
ports = "54321"

conn = ksycopg2.connect(
    f"dbname=TEST user=SYSTEM password=123456 host={hosts} port={ports}"
)

驅(qū)動(dòng)會(huì)按順序嘗試連接。第一個(gè)連不上，自動(dòng)切到第二個(gè)。

4.2 連接重試

集群主備切換時(shí)，數(shù)據(jù)庫服務(wù)會(huì)短暫不可用。配置重試參數(shù)可以避免應(yīng)用直接報(bào)錯(cuò)：

conn = ksycopg2.connect(
    f"dbname=TEST user=SYSTEM password=123456 host={hosts} port={ports} "
    "connect_timeout=3 retries=5 delay=3"
)

參數(shù)說明：

• retries=5：最多重試5次
• delay=3：每次重試間隔3秒
• connect_timeout=3：每次連接嘗試最長等3秒

這么配下來，數(shù)據(jù)庫最長需要等多久？實(shí)際邏輯是：第1次連接3秒超時(shí)，等3秒，第2次連接3秒超時(shí)，等3秒……主備切換通常10-15秒能完成，5次重試夠了。

4.3 自動(dòng)找主

集群環(huán)境里，只有主節(jié)點(diǎn)支持讀寫。可以配置target_session_attrs=read-write，讓驅(qū)動(dòng)自動(dòng)找到主節(jié)點(diǎn)連接：

conn = ksycopg2.connect(
    f"dbname=TEST user=SYSTEM password=123456 host={hosts} port={ports} "
    "connect_timeout=3 target_session_attrs=read-write"
)

驅(qū)動(dòng)會(huì)逐個(gè)主機(jī)嘗試，直到連到一個(gè)支持讀寫的節(jié)點(diǎn)。這個(gè)配置在讀寫分離場景下特別有用——應(yīng)用只管寫，不需要知道當(dāng)前誰是主。

4.4 快速故障轉(zhuǎn)移

fastswitch=on是另一個(gè)實(shí)用參數(shù)：

conn = ksycopg2.connect(
    f"dbname=TEST user=SYSTEM password=123456 host={hosts} port={ports} "
    "fastswitch=on retries=3 delay=5"
)

開啟后，驅(qū)動(dòng)會(huì)記住哪些主機(jī)掛了，后續(xù)連接請求自動(dòng)跳過故障主機(jī)，不用每次都等超時(shí)。對于短連接頻繁的場景，這個(gè)參數(shù)能明顯減少延遲。

參數(shù)對比：

4.5 連接負(fù)載均衡

如果后端有多個(gè)備節(jié)點(diǎn)，可以開啟負(fù)載均衡分散連接壓力：

conn = ksycopg2.connect(
    f"dbname=TEST user=SYSTEM password=123456 host={hosts} port={ports} "
    "loadbalance=on connect_timeout=6"
)

參數(shù)說明：

• loadbalance=on：開啟負(fù)載均衡
• 驅(qū)動(dòng)會(huì)在多個(gè)主機(jī)之間輪詢分配連接
• 適合讀多寫少、有多個(gè)備節(jié)點(diǎn)的場景

五、連接池

5.1 SimpleConnectionPool（單線程）

單線程場景（比如定時(shí)任務(wù)腳本），用SimpleConnectionPool：

import ksycopg2
from ksycopg2 import pool

connection_pool = ksycopg2.pool.SimpleConnectionPool(
    minconn=1,
    maxconn=10,
    database='TEST',
    user='SYSTEM',
    password='123456',
    host='127.0.0.1',
    port='54321'
)

# 獲取連接
conn = connection_pool.getconn()
cur = conn.cursor()
cur.execute("SELECT * FROM test_user")
rows = cur.fetchall()
cur.close()
# 放回連接池
connection_pool.putconn(conn)

# 關(guān)閉所有連接
connection_pool.closeall()

5.2 ThreadedConnectionPool（多線程）

多線程Web應(yīng)用（比如Flask、Django部署），建議用ThreadedConnectionPool：

connection_pool = ksycopg2.pool.ThreadedConnectionPool(
    minconn=5,
    maxconn=20,
    database='TEST',
    user='SYSTEM',
    password='123456',
    host='127.0.0.1',
    port='54321'
)

使用方法跟SimpleConnectionPool一樣。

一個(gè)需要注意的地方：putconn()只是把連接放回池子里，不會(huì)關(guān)閉連接。多線程環(huán)境下，如果多個(gè)線程同時(shí)拿到同一個(gè)連接，可能會(huì)出問題。需要自己在應(yīng)用層加鎖控制。

另一個(gè)常見的錯(cuò)誤：千萬不要對getconn()拿到的連接直接調(diào)用close()。那樣連接池里的連接數(shù)就少了一個(gè)，最后會(huì)報(bào)connection pool exhausted。

如果連接池滿了，getconn()會(huì)直接報(bào)錯(cuò)。解決辦法是調(diào)大maxconn，或者排查業(yè)務(wù)代碼有沒有及時(shí)putconn()。

5.3 完整的多線程示例

import threading
import ksycopg2
from ksycopg2 import pool

connection_pool = ksycopg2.pool.ThreadedConnectionPool(
    1, 5, database='TEST', user='SYSTEM',
    password='123456', host='127.0.0.1', port='54321'
)

def worker(n):
    conn = connection_pool.getconn()
    cur = conn.cursor()
    cur.execute("INSERT INTO test_user VALUES (%s, %s)", (n, f"user_{n}"))
    conn.commit()
    cur.close()
    connection_pool.putconn(conn)

threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

connection_pool.closeall()

六、自動(dòng)提交模式

ksycopg2默認(rèn)是手動(dòng)提交模式。執(zhí)行INSERT、UPDATE后必須調(diào)用commit()，否則數(shù)據(jù)不會(huì)真正寫入。

conn = ksycopg2.connect(database='TEST', user='SYSTEM')
conn.autocommit = False   # 默認(rèn)值
cur = conn.cursor()
cur.execute("INSERT INTO test_user VALUES (1, 'test')")
conn.commit()  # 必須調(diào)用

如果不想手動(dòng)提交，可以開啟自動(dòng)提交模式：

conn.autocommit = True

開啟后，每條語句執(zhí)行完立刻提交。適合原型開發(fā)，生產(chǎn)環(huán)境不建議用，事務(wù)控制會(huì)亂。

七、數(shù)據(jù)類型映射

Python類型和金倉數(shù)據(jù)庫類型的對應(yīng)關(guān)系：

需要注意的地方：

• Python 2里字符串是str和unicode，Python 3統(tǒng)一是str。
• 金倉的時(shí)間范圍比Python的datetime大，比如date最大到9999-12-31，Python也一樣，這個(gè)沒問題。
• bytea類型在Python 3里返回memoryview，需要用tobytes()轉(zhuǎn)換才能拿到原始字節(jié)。

示例：

cur.execute("SELECT bytea_column FROM test_lob")
row = cur.fetchone()
if isinstance(row[0], memoryview):
    data = row[0].tobytes().decode('utf-8')

八、編碼設(shè)置

連接時(shí)可以指定客戶端編碼，避免中文亂碼：

conn = ksycopg2.connect(
    database='TEST',
    user='SYSTEM',
    password='123456',
    client_encoding='UTF8'   # 指定編碼
)

也可以在連接后動(dòng)態(tài)設(shè)置：

conn.set_client_encoding('UTF8')

如果不指定，默認(rèn)用數(shù)據(jù)庫的編碼。建議統(tǒng)一用UTF8。

九、小結(jié)

上篇主要講了ksycopg2的安裝配置、連接管理、高可用和連接池?？偨Y(jié)幾個(gè)要點(diǎn)：

1. ksycopg2是Python連金倉的標(biāo)準(zhǔn)方式，支持Python 2.7到3.13
2. 安裝時(shí)注意依賴庫libkci的路徑配置
3. 高可用參數(shù)很重要：多主機(jī)、重試、自動(dòng)找主、快速故障轉(zhuǎn)移——這幾個(gè)配好了，集群切換對應(yīng)用影響小很多
4. 連接池用ThreadedConnectionPool，注意連接用完要putconn，不要close

以上就是Python連接管理金倉數(shù)據(jù)庫的完全指南的詳細(xì)內(nèi)容，更多關(guān)于Python連接金倉數(shù)據(jù)庫的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論