java - 如何确保我不会同时在两个线程之间共享同一个套接字？

Question

<分区>

Answer 1

首先，您需要一种方式让客户使用 Socket 通知您他们已完成.您可以添加一种允许他们发出信号的方法。这是合法的，它会起作用，但你必须依靠你的客户表现良好。或者更确切地说，使用您的套接字的程序员不会忘记返回它。有一种模式可以帮助解决这个问题:execute around图案。而不是给出 Socket ，你创建了一个接受 Consumer<Socket> 的方法，然后执行消费者，返回Socket本身。

public void useSocket(Consumer<Socket> socketUser) {
    Socket socket = getSocket();
    try {
        socketUser.accept(socket);
    } finally {
        returnSocket(socket);
    }
}

现在让我们看看我们将如何实现 getSocket()和 returnSocket() .显然，它涉及从某种集合中获取它们，并将它们返回到该集合中。 Queue在这里是一个不错的选择(正如其他人也指出的那样)。它允许从一侧获取它，并在另一侧返回，此外还有大量高效的线程安全实现，而且 taker 和 adder 通常不会相互争用。既然你事先知道 socket 的数量，我会选择 ArrayBlockingQueue .

这里的另一个问题是您的实现返回一个 Optional .我不确定如果没有可用的客户会做什么 Socket , 但如果它正在等待并重试，我建议你简单地制作 getSocket()阻塞在队列中。实际上，我会尊重您方法的这一方面，并​​考虑到可能没有 Socket。可用的。对于 execute around 方法，这会将其转换为 useSocket()方法返回 false如果没有 Socket可用。

private final BlockingQueue<Socket> queue;

public SocketPool(Set<Socket> sockets) {
    queue = new ArrayBlockingQueue<>(sockets.size());
    queue.addAll(sockets);
}

public boolean useSocket(Consumer<Socket> socketUser) throws InterruptedException {
    Optional<Socket> maybeSocket = getSocket();
    try {
        maybeSocket.ifPresent(socketUser);
        return maybeSocket.isPresent();
    } finally {
        maybeSocket.ifPresent(this::returnSocket);
    }
}

private void returnSocket(Socket socket) {
    queue.add(socket);
}

private Optional<Socket> getSocket() throws InterruptedException {
    return Optional.ofNullable(queue.poll());
}

就是这样，就是你的 SocketPool .

啊，但接下来是小气的一点:检查 active 。这很吝啬，因为您的 Activity 检查实际上与您的普通客户竞争。

为了解决这个问题，我建议如下:让您的客户报告是否 Socket他们得到的是活的还是不活的。由于检查 Activity 性归结为使用套接字，这对您的客户来说应该很简单。

所以不是 Consumer<Socket> ，我们将采用 Function<Socket, Boolean> .如果函数返回 false ，我们将考虑 Socket不再活着。在这种情况下，我们不会将其添加回常规队列，而是将其添加到死套接字集合中，并且我们将有一个计划任务，间歇性地重新检查死套接字。由于这是在单独的集合上发生的，因此计划检查不再与常规客户竞争。

现在你可以制作一个 SocketManager用Map将数据中心映射到 SocketPool实例。此映射不需要更改，因此您可以将其设为最终并在 SocketManager 中对其进行初始化。的构造函数。

这是我对 SocketPool 的初步代码(未经测试):

class SocketPool implements AutoCloseable {

    private final BlockingQueue<Socket> queue;
    private final Queue<Socket> deadSockets = new ConcurrentLinkedQueue<>();
    private final ScheduledFuture<?> scheduledFuture;

    public SocketPool(Set<Socket> sockets, ScheduledExecutorService scheduledExecutorService) {
        queue = new ArrayBlockingQueue<>(sockets.size());
        queue.addAll(sockets);
        scheduledFuture = scheduledExecutorService.scheduleAtFixedRate(this::recheckDeadSockets, 60, 60, TimeUnit.SECONDS);
    }

    public boolean useSocket(Function<Socket, Boolean> socketUser) throws InterruptedException {
        Optional<Socket> maybeSocket = getSocket();
        boolean wasLive = true;
        try {
            wasLive = maybeSocket.map(socketUser).orElse(false);
            return wasLive && maybeSocket.isPresent();
        } finally {
            boolean isLive = wasLive;
            maybeSocket.ifPresent(socket -> {
                if (isLive) {
                    returnSocket(socket);
                } else {
                    reportDead(socket);
                }
            });
        }
    }

    private void reportDead(Socket socket) {
        deadSockets.add(socket);
    }

    private void returnSocket(Socket socket) {
        queue.add(socket);
    }

    private Optional<Socket> getSocket() throws InterruptedException {
        return Optional.ofNullable(queue.poll());
    }

    private void recheckDeadSockets() {
        for (int i = 0; i < deadSockets.size(); i++) {
            Socket socket = deadSockets.poll();
            if (checkAlive(socket)) {
                queue.add(socket);
            } else {
                deadSockets.add(socket);
            }
        }
    }

    private boolean checkAlive(Socket socket) {
        // do actual live check with SendSocket class, or implement directly in this one
        return true;
    }

    @Override
    public void close() throws Exception {
        scheduledFuture.cancel(true);
    }
}