java - 避免在对象上使用同步锁

Question

我在 SocketManager 类中有以下方法，该方法由后台线程每 60 秒调用一次。它将 ping 套接字并检查它是否处于 Activity 状态，并将所有内容放入 liveSocketsByDatacenter 映射中。

  private final Map<Datacenters, List<SocketHolder>> liveSocketsByDatacenter =
      new ConcurrentHashMap<>();

  // runs every 60 seconds to ping all the socket to make sure whether they are alive or not
  private void updateLiveSockets() {
    Map<Datacenters, List<String>> socketsByDatacenter = Utils.SERVERS;

    for (Map.Entry<Datacenters, List<String>> entry : socketsByDatacenter.entrySet()) {
      List<SocketHolder> liveSockets = liveSocketsByDatacenter.get(entry.getKey());
      List<SocketHolder> liveUpdatedSockets = new ArrayList<>();
      for (SocketHolder liveSocket : liveSockets) {
        Socket socket = liveSocket.getSocket();
        String endpoint = liveSocket.getEndpoint();
        Map<byte[], byte[]> holder = populateMap();
        Message message = new Message(holder, Partition.COMMAND);

        // pinging to see whether a socket is live or not
        boolean status = SendToSocket.getInstance().execute(message.getAdd(), holder, socket);
        boolean isLive = (status) ? true : false;

        SocketHolder zmq = new SocketHolder(socket, liveSocket.getContext(), endpoint, isLive);
        liveUpdatedSockets.add(zmq);
      }
      liveSocketsByDatacenter.put(entry.getKey(), Collections.unmodifiableList(liveUpdatedSockets));
    }
  }

此外，我在同一个 SocketManager 类中还有以下方法。 getNextSocket() 方法将由多个读取器线程(假设最多 10 个线程)同时调用，以获取下一个 Activity 套接字。

  // this method will be called by multiple threads concurrently to get the next live socket
  public Optional<SocketHolder> getNextSocket() {
    for (Datacenters dc : Datacenters.getOrderedDatacenters()) {
      Optional<SocketHolder> liveSocket = getLiveSocket(liveSocketsByDatacenter.get(dc));
      if (liveSocket.isPresent()) {
        return liveSocket;
      }
    }
    return Optional.absent();
  }

  private Optional<SocketHolder> getLiveSocket(final List<SocketHolder> listOfEndPoints) {
    if (!listOfEndPoints.isEmpty()) {
      // The list of live sockets
      List<SocketHolder> liveOnly = new ArrayList<>(listOfEndPoints.size());
      for (SocketHolder obj : listOfEndPoints) {
        if (obj.isLive()) {
          liveOnly.add(obj);
        }
      }
      if (!liveOnly.isEmpty()) {
        // The list is not empty so we shuffle it an return the first element
        return Optional.of(liveOnly.get(random.nextInt(liveOnly.size()))); // just pick one
      }
    }
    return Optional.absent();
  }

问题陈述:

我想确保所有这 10 个线程 + 计时器线程在调用 getNextSocket() 方法后永远不会使用相同的套接字。

• 如果计时器线程正在 updateLiveSockets() 方法中的 socketA 上工作，那么所有这 10 个线程都应该在其他 Activity 套接字上工作(这 10 个线程中的每一个都在不同的 Activity 套接字上工作)套接字)
• 所有这 10 个线程应该始终在不同的 Activity 套接字上工作。
• 此外，如果我们的读取器线程多于可用的实时套接字，那么我仍然需要确保每个读取器线程在不同的实时套接字上工作。一旦一个读取器线程使用 Activity 套接字完成，其他读取器线程就可以立即使用该套接字或使用任何更好的方法。我想尽可能减少阻塞时间。

解决这个问题的最佳方法是什么？我可以在这 10 个读取器线程以及计时器线程中对套接字进行同步，这将保证只有线程在该套接字上工作，但我不想在这里使用同步。必须有更好的方法来确保每个线程同时使用不同的单个 Activity 套接字，而不是在特定套接字上进行同步。我有大约 60 个套接字和大约 10 个读取器线程加上 1 个计时器线程。我需要在这里使用 ThreadLocal 概念吗？

Answer 1

解决您的问题的最佳解决方案是使用 ConcurrentQueue您不需要使用ThreadLocal。 ConcurrentQueue 是非阻塞的，对于多线程环境非常高效。例如，这就是如何删除不活动的套接字并保留 Activity 的套接字。

    private final Map<Datacenters, ConcurrentLinkedQueue<SocketHolder>> liveSocketsByDatacenter =
          new ConcurrentHashMap<>();


// runs every 60 seconds to ping 70 sockets the socket to make sure whether they are alive or not (it does not matter if you ping more sockets than there are in the list because you are rotating the que)
  private void updateLiveSockets() {
    Map<Datacenters, List<String>> socketsByDatacenter = Utils.SERVERS;

    for (Map.Entry<Datacenters, List<String>> entry : socketsByDatacenter.entrySet()) {
    Queue<SocketHolder> liveSockets = liveSocketsByDatacenter.get(entry.getKey());
      for (int i = 0; i<70; i++) {
            SocketHolder s = liveSockets.poll();
        Socket socket = s.getSocket();
        String endpoint = s.getEndpoint();
        Map<byte[], byte[]> holder = populateMap();
        Message message = new Message(holder, Partition.COMMAND);

        // pinging to see whether a socket is live or not
        boolean status = SendToSocket.getInstance().execute(message.getAdd(), holder, socket);
        boolean isLive = (status) ? true : false;

        SocketHolder zmq = new SocketHolder(socket, s.getContext(), endpoint, isLive);
        liveSockets.add(zmq);
      }
    }
  }