Java 可调用线程 : keep configuration

Question

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

澄清一下:

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

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

我正在寻找保持连接处于 Activity 状态并为每个线程重复使用它们的最佳实践。

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

• 在需要时使用线程池内的连接池检索空闲连接(造成死锁和其他线程安全问题)
• 使用带有连接和 using the thread number 的静态(或类似)数组检索其连接(也不是防污，请参阅链接)

实现此目的最有效的方法是什么？ <- 老问题，请参阅新问题的编辑。

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

map<String/threadId, connection>

并将整个 map (引用)传递给可调用的。这样我可以使用类似的东西:(伪代码)

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

也可以将 map 设为静态并静态访问它。这样我就不必将 map 传递给 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将新的查询添加到发件箱，如果这引发异常，则您的队列已满，您需要降低查询创建率和/或创建更多工作线程。要打破工作线程的无限循环，请在其 Future 上调用 cancel(true)，这将在 Worker 内部抛出 InterruptedException .