java - 在服务于 Socket 连接的两个线程之间共享公共(public)数据

Question

我在 SO 上看到很多类似的问题，但几乎没有一个在图片中有 Socket 。所以请花点时间阅读问题。

我有服务器应用程序(使用 ServerSocket)监听请求，当客户端尝试连接时，会创建新线程来为客户端提供服务(并且服务器返回监听模式以接收新请求).现在，我需要根据其他客户端发送到服务器的内容来响应一个客户端。

示例:

• ServerSocket 监听传入连接。
• 客户端 A 连接，创建新线程为 A 服务。
• 客户端 B 连接，创建新线程为 B 服务。
• A 向服务器发送消息“Hello from A”。
• 将此消息作为对客户端 B 的响应发送。

我是整个“线程间通信”的新手。显然，上面提到的情况听起来很简单，但我描述这个是为了获得提示，因为我将在客户端之间交换大量数据，将服务器保持为中间。

此外，如果我想将共享对象限制在例如 10 个特定客户端，该怎么办？这样，当第 11 个客户端连接到服务器时，我创建了新的共享对象，它将用于在第 11、12、13 ...... 到第 20 个客户端之间交换数据。对于每组 10 个客户端，依此类推。

我尝试了什么:(我猜是愚蠢的)

• 我有一个 public 类，该类的对象应该作为 public static 共享，这样我就可以将它用作全局对象而无需实例化它，例如 MyGlobalClass .SharedMsg.
• 那行不通，我无法将一个线程中收到的数据发送到另一个线程。

我知道有一个明显的锁定问题，因为如果一个线程正在写入一个对象，则在第一个线程完成写入之前，其他线程无法访问它。

那么解决这个问题的理想方法是什么？

更新

由于我为传入连接请求创建线程的方式，我无法理解如何在线程之间共享相同的对象，因为如上所述使用全局对象不起作用。

以下是我如何监听传入连接并动态创建服务线程。

// Method of server class
public void startServer()
{
    if (!isRunning)
    {
        try
        {
            isRunning = true;
            while (isRunning)
            {
                try
                {
                    new ClientHandler(mysocketserver.accept()).start();
                }
                catch (SocketTimeoutException ex)
                {
                    //nothing to perform here, go back again to listening.
                }
                catch (SocketException ex)
                {
                    //Not to handle, since I'll stop the server using SocketServer's close() method, and its going to throw SocketException anyway.
                }
            }
        }
        catch (Exception ex)
        {
            ex.printStackTrace();
        }
    }
    else
        System.out.println("Server Already Started!");
}

还有 ClientHandler 类。

public class ClientHandler extends Thread
{
    private Socket client = null;
    private ObjectInputStream in = null;
    private ObjectOutputStream out = null;

    public ClientHandler(Socket client)
    {
        super("ClientHandler");
        this.client = client;
    }

    //This run() is common for every Client that connects, and that's where the problem is.
    public void run()
    {
        try
        {
            in = new ObjectInputStream(client.getInputStream());
            out = new ObjectOutputStream(client.getOutputStream());

            //Message received from this thread.
            String msg = in.readObject().toString();
            System.out.println("Client @ "+ client.getInetAddress().getHostAddress() +" Says : "+msg);


            //Response to this client.
            out.writeObject("Message Received");

            out.close();
            in.close();
            client.close();
        }
        catch (Exception ex)
        {
            ex.printStackTrace();
        }
    }
}

我相信我创建动态线程来服务每个连接的客户端的方式，使用 Global 对象共享相同的数据源是不可能的，因为上面的 run() 的主体正是对于每个连接的客户端都是一样的，因此这个相同的方法既是消费者又是生产者。我应该进行哪些修复才能为每个连接创建动态线程并仍然共享同一个对象。

Answer 1

您可能需要一个队列用于每个客户端之间的通信。每个队列都是将数据从一个客户端推送到另一个客户端的“管道”。

你会像这样使用它(伪代码):

Thread 1:
Receive request from Client A, with message for Client B
Put message on back of concurrent Queue A2B
Respond to Client A.

Thread 2:
Receive request from Client B.
Pop message from front of Queue A2B
Respond to Client B with message.

您可能还希望它是通用的，因此您有一个 AllToB 队列，许多客户端(因此有许多线程)可以写入。

注意类别:ConcurrentLinkedQueue , ArrayBlockingQueue .

如果你想限制消息的数量，ArrayBlockingQueue 及其容量构造函数可以让你做到这一点。如果不需要阻塞功能，可以使用方法 offer 和 poll 而不是 put 和 take.

我不担心共享队列，这会使问题变得更加复杂。仅当您知道需要解决内存使用问题时才这样做。

编辑:根据您的更新:

如果您需要在所有动态创建的实例之间共享一个实例，您可以:

1. 创建一个静态实例。
2. 将其传递给构造函数。

示例 1:

public class ClientHandler extends Thread
{
  public static final Map<ClientHandler, BlockingQueue<String>> messageQueues 
    = new ConcurrentHashMap<>();

  <snip>

  public ClientHandler(Socket client)
  {
    super("ClientHandler");
    this.client = client;
    // Note: Bad practice to reference 'this' in a constructor.
    // This can throw an error based on what the put method does.
    // As such, if you are to do this, put it at the end of the method.
    messageQueues.put(this, new ArrayBlockingQueue<>());
  }

  // You can now access this in the run() method like so:
  // Get messages for the current client.
  // messageQueues.get(this).poll();
  // Send messages to the thread for another client.
  // messageQueues.get(someClient).offer(message);

一些注意事项:

• messageQueues 对象实际上应该包含某种客户端标识符，而不是短暂的对象引用。
• 更可测试的设计会将 messageQueues 对象传递到构造函数中以允许模拟。
• 我可能会建议为 map 使用包装器类，这样您就可以使用 2 个参数调用 offer，而不必担心 map 语义。