如何在Python独立进程间共享SQL连接并传递查询结果？

你面临的核心问题是：跨独立Shell进程（a.py启动b/c.py）共享SQL连接并动态传输大小不确定的查询结果，直接通过内存地址访问跨进程Python对象是不可行的——因为Python对象的内存布局依赖解释器内部状态（如GC、引用计数），跨进程直接读取内存地址会导致数据损坏或程序崩溃。

以下是几种可靠的替代方案：

方案1：流式Socket通信（手动处理数据长度） #

通过Socket先传输结果的字节长度，再发送序列化后的内容，无需预先指定固定大小，完美适配从KB到百MB级的结果。

a.py（服务端，处理查询） #

import socket
import pickle
import sqlite3
import struct

# 初始化SQL连接
conn = sqlite3.connect('test.db')
cur = conn.cursor()

# 启动Socket服务
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.bind(('localhost', 9999))
server_sock.listen(5)

print("等待客户端连接...")
while True:
    client_sock, addr = server_sock.accept()
    # 接收查询指令
    query = client_sock.recv(4096).decode('utf-8')
    if not query:
        client_sock.close()
        continue
    
    # 执行查询并序列化结果
    cur.execute(query)
    results = cur.fetchall()
    serialized_data = pickle.dumps(results)
    
    # 先发送结果字节长度（打包为8字节无符号长整型）
    client_sock.sendall(struct.pack('!Q', len(serialized_data)))
    # 发送完整结果数据
    client_sock.sendall(serialized_data)
    
    client_sock.close()

b.py（客户端，发起查询） #

import socket
import pickle
import struct

client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_sock.connect(('localhost', 9999))

# 发送查询请求
query = "select * from test_table"
client_sock.sendall(query.encode('utf-8'))

# 接收结果长度
length_data = client_sock.recv(8)
if not length_data:
    print("无返回数据")
    client_sock.close()
    exit()
data_length = struct.unpack('!Q', length_data)[0]

# 分块接收完整结果
received_data = b''
while len(received_data) < data_length:
    chunk = client_sock.recv(min(4096, data_length - len(received_data)))
    if not chunk:
        break
    received_data += chunk

# 反序列化并打印结果
results = pickle.loads(received_data)
print(results)

client_sock.close()

方案2：RPC框架（Python自带，实现极简） #

使用Python内置的XML-RPC框架，自动处理数据序列化、传输和大小适配，无需手动处理Socket细节。

a.py（RPC服务端） #

from xmlrpc.server import SimpleXMLRPCServer
import sqlite3

# 初始化SQL连接
conn = sqlite3.connect('test.db')
cur = conn.cursor()

def execute_sql_query(query):
    cur.execute(query)
    return cur.fetchall()

# 启动RPC服务
server = SimpleXMLRPCServer(('localhost', 8000))
server.register_function(execute_sql_query, 'execute_query')
print("RPC服务启动，监听端口8000")
server.serve_forever()

b.py（RPC客户端） #

from xmlrpc.client import ServerProxy

# 连接RPC服务
proxy = ServerProxy('http://localhost:8000/')
# 发起查询并获取结果
results = proxy.execute_query("select * from test_table")
print(results)

方案3：共享内存动态扩容（不推荐） #

如果一定要用共享内存，可通过"先传大小+动态重建共享内存"的方式实现，但逻辑复杂且容易出问题：

1. a进程创建初始共享内存，b进程连接
2. a执行查询后，若结果超过当前共享内存大小，则创建新的更大的共享内存，将结果写入
3. a通过Socket/管道将新的共享内存名称和大小发送给b
4. b断开旧共享内存，连接新共享内存读取数据

这种方式的可靠性远低于前两种方案，仅在特殊场景下考虑使用。

内容的提问来源于stack exchange，提问作者Scott Paterson