方法介绍

以下内容介绍一个简单的固定任务示例。

首先，创建一个最大连接数为8的连接池 DBConnectionPool。和通常连接池有所不同，当不再使用连接时，API 不会立刻销毁该连接，而是直到析构 DBConnectionPool 时才进行销毁，或者手动执行 shutDown() 关闭 DBConnectionPool 时才会销毁连接。

import dolphindb as ddb
import time
import asyncio

pool = ddb.DBConnectionPool("localhost", 8848, 8)

创建一个协程任务函数，使用 sleep 模拟一段运行的时间。

async def test_run(i):
    try:
        return await pool.run(f"sleep(2000);1+{i}")
    except Exception as e:
        print(e)

定义任务列表，并创建一个事件循环对象，运行任务列表直到完成全部任务。

tasks = [
    asyncio.ensure_future(test_run(1)),
    asyncio.ensure_future(test_run(3)),
    asyncio.ensure_future(test_run(5)),
    asyncio.ensure_future(test_run(7)),
]

loop = asyncio.get_event_loop()
try:
    time_st = time.time()
    loop.run_until_complete(asyncio.wait(tasks))
    time_ed = time.time()
except Exception as e:
    print("catch e:")
    print(e)

任务结束后，打印执行时间和各个任务的结果，并关闭连接池对象。

print("time: ", time_ed-time_st)

for task in tasks:
    print(task.result())

pool.shutDown()

期望的输出结果如下所示：

time:  2.0017542839050293
2
4
6
8

上述例子展示了已固定脚本任务调用 DBConnectionPool 的用法，在 Python 中只有一个主线程，但使用了协程创建子任务并调用 DBConnectionPool 以实现运行。须注意，实际上 Python API 的底层实现是通过使用 C++ 线程以维护每一个连接。若提交任务数超出实际线程数，则可能出现任务迟迟没有执行的情况，与通常的协程并发有一定区别。

此外，用户也可以自定义传入脚本的对象，可参考下述示例2。

下例定义了一个可以传入自定义脚本作为参数的类，并配合 Python 的多线程机制动态添加子任务。

import dolphindb as ddb
import time
import asyncio
import threading

# 在该例子中主线程负责创建协程对象传入自定义脚本并调用自定义的对象去运行，并新起子线程运行事件循环防止阻塞主线程。
class DolphinDBHelper(object):
    pool = ddb.DBConnectionPool("localhost", 8848, 10)
    @classmethod
    async def test_run(cls,script):
        print(f"run script: [{script}]")
        return await cls.pool.run(script)

    @classmethod
    async def runTest(cls,script):
        start = time.time()
        task = loop.create_task(cls.test_run(script))
        result = await asyncio.gather(task)
        print(f"""[{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}] time: {time.time()-start} result: {result}""")
        return result

# 定义一个跑事件循环的线程函数
def start_thread_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

if __name__=="__main__":
    start = time.time()
    print("In main thread",threading.current_thread())
    loop = asyncio.get_event_loop()
    # 在子线程中运行事件循环, 让它 run_forever
    t = threading.Thread(target= start_thread_loop, args=(loop,))
    t.start()
    task1 = asyncio.run_coroutine_threadsafe(DolphinDBHelper.runTest("sleep(1000);1+1"),loop)
    task2 = asyncio.run_coroutine_threadsafe(DolphinDBHelper.runTest("sleep(3000);1+2"),loop)
    task3 = asyncio.run_coroutine_threadsafe(DolphinDBHelper.runTest("sleep(5000);1+3"),loop)
    task4 = asyncio.run_coroutine_threadsafe(DolphinDBHelper.runTest("sleep(1000);1+4"),loop)

    end = time.time()
    print("main thread time: ", end - start)

运行结果如下：

In main thread <_MainThread(MainThread, started 139838803788160)>
main thread time:  0.00039839744567871094
run script: [sleep(1000);1+1]
run script: [sleep(3000);1+2]
run script: [sleep(5000);1+3]
run script: [sleep(1000);1+4]
[2023-03-14 16:46:56] time: 1.0044968128204346 result: [2]
[2023-03-14 16:46:56] time: 1.0042989253997803 result: [5]
[2023-03-14 16:46:58] time: 3.0064148902893066 result: [3]
[2023-03-14 16:47:00] time: 5.005709409713745 result: [4]

上述例子中，在主线程中创建子线程开启事件循环，并指定该事件循环一直保持运行。随后向该事件循环中加入四个脚本执行任务，每个任务分别需要耗时 1s、3s、5s和1s。从主线程打印 main thread time: 0.00039839744567871094 可以看出，四个事件放入事件循环后实现了异步执行，随后每个协程都打印自身执行的结束时间和时长。由于任务1和任务4耗时一致，因此同时打印结果；2s后任务2执行结束；再过2s后任务3也执行结束。由结果可知，四个任务的执行结果符合并发执行的预期。

addTask(script, taskId, *args, **kwargs)

• script：待执行的 DolphinDB 脚本。
• taskId：指定的任务 Id。
• *args：传递给 DolphinDB 函数的参数。
• **kwargs：
  • clearMemory：是否查询后释放变量。True 表示释放，默认值为 True。
  • pickleTableToList：是否将结果的 Table 类型对象转换为 list 类型对象。True 表示转换为 list 类型对象，False 表示转换为 DataFrame 类型对象，默认值为 False。

addTask 根据 taskId 将任务提交至 DBConnectionPool 的连接池中，由连接池分配连接执行脚本任务。如下所示，调用 addTask 向连接池中添加一个 taskId 为 12 的任务。

pool.addTask("sleep(1000);1+2", taskId=12)

isFinished(taskId)

• taskId：查询的任务 Id。

该函数通过 taskId 来查询对应任务是否已经完成。如果已完成，则返回 True；反之返回 False。

简单使用示例如下：

if pool.isFinished(taskId=12):
    print("task has done!")

getData(taskId)

• taskId：查询的任务 Id。

该函数通过 taskId 来查询对应任务的返回结果。

简单使用示例如下：

res = pool.getData(taskId=12)

注意 ： 每次执行 addTask 指定 taskId 并创建任务后，使用 getData 方法只能对该 taskID 对应任务的返回结果执行一次查询。若在创建任务后未调用 getData 方法，则在下次使用 addTask 指定同一 taskId 并创建任务时，其执行结果将覆盖掉前一次该 taskId 对应任务的执行结果。

在如下脚本中，首先创建一个 DBConnectionPool 连接池对象，然后调用 addTask 向连接池中添加一个 taskId 为12的任务，随后通过 isFinished 方法判断任务是否执行完毕，执行完毕后跳出循环，调用 getData 方法获取任务结果。

import dolphindb as ddb
import time

pool = ddb.DBConnectionPool("localhost", 8848, 8)
taskid = 12
pool.addTask("sleep(1500);1+2", taskId=taskid)
while True:
    if pool.isFinished(taskId=taskid):
        break
    time.sleep(0.01)

res = pool.getData(taskId=taskid)
print(res)

// output:
3

pool.shutDown()

该方法用于关闭不再使用的 DBConnectionPool，停止线程池中使用的事件循环，并且中止所有的异步任务。调用 shutDown 方法后，关闭的线程池不可继续使用。

注意 ： 如果使用了 runTaskAsync() 的方式创建异步任务，必须在不使用 DBConnectionPool 时调用该函数。

sessionids = pool.getSessionId()

该方法用于获得当前线程池中所有 session 会话中的 session Id。