Java 可调用线程：保持配置

Question

我正在设置一个服务器 (Radius) 的模拟器（用于测试），它使用线程向另一台服务器 (LDAP) 发送查询。 查询需要每秒 x 次执行。 为此，我正在使用带有可调用对象的计划线程池执行程序，以便我可以创建可调用对象并将它们提交到线程池以执行。 每个线程都应该打开自己的连接并使用它来查询。 问题是我希望每次使用连接时都能被同一个线程重新使用。

澄清一下：

如果我有一个 20 个线程池，我希望创建和使用 20 个连接。 （所以我可以发送 10.000 个查询，这些查询将由 20 个线程/连接依次处理）。

现在，要连接的 (LDAP) 服务器信息被发送到可调用对象的构造函数，并且可调用对象设置连接以供执行。此后，我使用未来的可调用系统检索结果。 问题是每次我创建一个可调用连接时，连接都会被打开（当然后来会关闭）。

我正在寻找最佳实践，以保持连接处于活动状态，并将它们重新用于每个线程。

我已经想到了一些方法来实现这一点，但它们似乎效率不高：

• 在需要时使用我的线程池中的连接池来检索空闲连接（造成死锁和其他线程安全问题）
• 使用带有连接和using the thread number 的静态（或左右）数组来检索其连接（也不是防犯规，请参阅链接）

什么是最有效的实现方式？

编辑： 我在想，因为我无法安全地获得线程号，但 threadId 始终是唯一的，我可以只使用一个

map<String/threadId, connection>

并将整个地图（参考）传递给可调用对象。这样我可以使用类似的东西：（伪代码）

Connection con = map.get(this.getThreadId());
If (con == null){
  con = new Connection(...);
  map.put(this.getThreadId(), con)
}

也可以将地图设为静态并仅静态访问它。这样我就不必将地图传递给 Callable。 这至少是安全的，不会强迫我重构我的代码。

新问题： 什么会更符合最佳实践；上述解决方案还是 Zim-Zam 的解决方案？ 如果以上是最好的，选择静态解决方案会更好还是不？

Answer 1

我会使用Callables 之间共享的BlockingQueue 来实现这一点，ScheduledThreadPoolExecutor 每秒将x 查询放入BlockingQueue 中

public class Worker implements Runnable {
    private final BlockingQueue<Query> inbox;
    private final BlockingQueue<Result> outbox;

    public Worker(BlockingQueue<Query> inbox, BlockingQueue<Result> outbox) {
        // create LDAP connection
        this.inbox = inbox;
        this.outbox = outbox;
    }

    public void run() {
        try {
            while(true) {
                // waits for a Query to be available
                Query query = inbox.take();
                // execute query
                outbox.add(new Result(/* result */));
            }
        } catch(InterruptedException e) {
          // log and restart? close LDAP connection and return?
        }
    }
}

public class Master {
   private final int x; // number of queries per second
   private final BlockingQueue<Query> outbox = new ArrayBlockingQueue<>(4 * x);
   private final BlockingQueue<Result> inbox = new ArrayBlockingQueue<>(4 * x);
   private final ScheduledThreadPoolExecutor executor;
   private final List<Future<?>> workers = new ArrayList<>(20);
   private final Future<?> receiver;

   public Master() {
     // initialize executor
     for(int i = 0; i < 20; i++) {
         Worker worker = new Worker(inbox, outbox);
         workers.add(executor.submit(worker));
     }

     receiver = executor.submit(new Runnable() {
         public void run() {
           while(!Thread.interrupted()) {
             try {
               Result result = inbox.take();
               // process result
             } catch(InterruptedException e) {
               return;
             }
           }
         }
     }
   }

   executor.scheduleWithFixedDelay(new Runnable() {
       public void run() {
           // add x queries to the queue
       }
   }, 0, 1, TimeUnit.SECONDS);
}

使用BlockingQueue#add 将新的Queries 添加到outbox，如果这引发异常，则您的队列已满，您需要降低查询创建速度和/或创建更多工作人员。要在其Future 上打破工人的无限循环调用cancel(true)，这将在Worker 内抛出一个InterruptedException。