使用 `oracledb` 模块在 Python 中建立与 Oracle 数据库的连接，可以采用多种方式，包括独立连接模式和连接池模式。以下是详细的连接方法说明。 ### 使用独立连接模式 独立连接模式适用于简单的应用场景，每次连接数据库时都会创建一个新的连接。以下是使用 `oracledb` 模块建立独立连接的示例代码： ```python import oracledb # 建立连接 connection = oracledb.connect( user='your_username', password='your_password', dsn='your_host:your_port/your_service_name' ) # 创建游标并执行查询 cursor = connection.cursor() cursor.execute("SELECT * FROM your_table") # 获取查询结果 rows = cursor.fetchall() for row in rows: print(row) # 关闭游标和连接 cursor.close() connection.close() ``` 在上述代码中，`dsn` 参数的格式为 `host:port/service_name`，其中 `host` 是 Oracle 数据库服务器的地址，`port` 是监听端口，默认为 1521，`service_name` 是数据库的服务名[^2]。 ### 使用连接池模式 对于需要频繁连接数据库的应用场景，使用连接池可以有效减少连接创建和销毁的开销，提高性能。以下是使用 `oracledb` 模块建立连接池的示例代码： ```python import oracledb # 创建连接池 pool = oracledb.create_pool( user='your_username', password='your_password', dsn='your_host:your_port/your_service_name', min=1, max=5, increment=1 ) # 从连接池中获取连接 connection = pool.acquire() # 创建游标并执行查询 cursor = connection.cursor() cursor.execute("SELECT * FROM your_table") # 获取查询结果 rows = cursor.fetchall() for row in rows: print(row) # 关闭游标并释放连接回池 cursor.close() pool.release(connection) # 关闭连接池 pool.close() ``` 在上述代码中，`create_pool` 函数用于创建连接池，`min`、`max` 和 `increment` 参数分别指定连接池中的最小连接数、最大连接数和每次增加的连接数。`acquire` 方法用于从连接池中获取一个连接，`release` 方法用于将连接释放回连接池，`close` 方法用于关闭整个连接池[^2]。 ### 连接超时处理 在实际应用中，可能会遇到数据库连接超时的问题。为了更好地处理这种情况，可以在连接时加入超时处理机制。以下是一个简单的示例代码： ```python import oracledb import threading import queue def run_query(sql, result_queue): try: # 建立连接 connection = oracledb.connect( user='your_username', password='your_password', dsn='your_host:your_port/your_service_name' ) # 创建游标并执行查询 cursor = connection.cursor() cursor.execute(sql) # 获取查询结果 rows = cursor.fetchall() result_queue.put(rows) # 关闭游标和连接 cursor.close() connection.close() except Exception as e: result_queue.put(e) def run_sql_oracle(sql: str): result_queue = queue.Queue() query_thread = threading.Thread(target=run_query, args=(sql, result_queue)) query_thread.start() # 为查询设置一个超时限制 timeout = 15 # 15秒 try: # 尝试在超时时间内获取结果 results = result_queue.get(timeout=timeout) # 检查结果是否为异常，如果是则进行相应处理 if isinstance(results, Exception): print(f"查询失败: {results}") return f"{results}" else: return results except queue.Empty: print("查询超时。") return "查询超时。" finally: query_thread.join() ``` 在上述代码中，`run_sql_oracle` 函数通过多线程执行数据库查询，并设置了一个超时限制。如果在指定时间内未能获取查询结果，则认为查询超时，并返回相应的提示信息[^5]。 ###