使用多个线程和连接池时避免连接超时

I'm splitting up a readonly database operation into multiple chunks, (Each of which reads a subset of a very large amount of data, analyzes it and writes the results to a disk file).

我正在将一个只读数据库操作分成多个块(每个块读取一个非常大量数据的子集,分析它并将结果写入磁盘文件)。

Each chunk performs a Select (into a datatable) on a new .net thread (using a delegate and BeginInvoke)

每个块在新的.net线程上执行Select(进入数据表)(使用委托和BeginInvoke)

There are more subsets of the data than there are available connections in the pool, so when I run out of connections, before the first one is released, subsequent connection requests are queued up... until the connection timeout expires, and then I get a timeout exception.

数据的子集比池中的可用连接多,所以当我用完连接时,在第一个连接释放之前,后续连接请求排队等待......直到连接超时到期,然后我得到超时异常。

How do I either, A) inhibit the timeout connection exception when the connections in the pool are all in use, or B) detect that they are all in use before I even ask for another one so I can wait until one is available before asking?

我如何,A)当池中的连接全部被使用时禁止超时连接异常,或者B)在我甚至要求另一个之前检测到它们全部被使用,所以我可以等到一个可用之后再询问?

1 个解决方案

#1

Two solutions:

A) Configure your connection pool with a timeout of several days. That will block the pending tasks until a connection is returned. Drawback: This won't work when a task hangs.

A)使用几天的超时配置连接池。这将阻止挂起的任务,直到返回连接。缺点:当任务挂起时,这将不起作用。

B) Use a thread pool and a work queue. The thread pool must have the same size as the connection pool (i.e. one connection per thread). Put all the work in the queue and have the tasks fetch work items from the queue until the queue is empty.

B)使用线程池和工作队列。线程池必须与连接池具有相同的大小(即每个线程一个连接)。将所有工作放入队列并让任务从队列中获取工作项,直到队列为空。

Pseudocode for solution B:

解决方案B的伪代码:

public class Setup
    connPool = createConnectionPool(100);
    queue = createWorkQueue();
    putAllWorkItemsInQueue(queue);
    for (int i=0; i<connPool.size(); i++) {
        t = new WorkerThread(queue)
        list.add(t);
        t.start();
    }
    while (queue.size() != 0) {
        Thread.sleep(1000);
    }
    for (thread in list) {
        thread.interrupt();
    }

public class WorkerThread
    run() {
        while (true) {
            try {
                workUnit = queue.get(); // This blocks
                process(workUnit);
            } catch (InterruptedException e) {
                break;
            }
        }
    }

#1