Java可调用线程：保持配置

Question

我正在设置一个服务器（Radius）的模拟器（用于测试），它使用线程将查询发送到另一个服务器（LDAP）。查询需要以每秒x的速度执行。为此，我使用了一个带有callable的调度线程池执行器，这样我就可以创建callable并将它们提交给线程池执行。每个线程都应该打开自己的连接并使用它进行查询。问题是，每次使用连接时，我都希望相同的线程重复使用它。

澄清：

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

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

我正在寻找最佳实践，以保持连接活着，并为每个线程重新使用它们。

我想了一些方法来实现这一点，但它们似乎不是很有效：

• 使用我的线程池中的连接池在需要时检索空闲连接（造成死锁和其他线程安全问题）

实现这一点的最有效方法是什么？

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

map<String/threadId, connection>

并将整个映射（引用）传递给可调用对象。这样我就可以使用：（伪代码）

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

也可以使地图保持静态，并仅以静态方式访问它。这样我就不必将映射传递给可调用对象。这至少是安全的，不会强迫我重新构造代码。

新问题：什么更符合最佳实践；上述解决方案还是齐姆-扎姆的解决方案？如果以上是最好的，那么静态解决方案会更好吗？

Answer 1

我将使用BlockingQueue来实现这一点，它在Callable之间共享，并且每隔一秒钟将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向发件箱添加新的查询，如果这引发异常，那么您的队列已满，您需要降低查询创建速率和/或创建更多工作人员。要在其未来上中断工作进程的无限循环调用cancel（true），这将在工作进程内部抛出一个中断异常。