JAVA套接字（Socket）101七天系列—第六天【一个带有连接池的示例】

时间：2023-01-16 20:43:10
1. 介绍

我们现在已经拥有的 MultithreadedServer 每当有客户机申请一个连接时都在一个新 Thread 中创建一个新ConnectionHandler。这意味着可能有一捆 Thread “躺”在我们周围。而且创建 Thread 的系统开销并不是微不足道的。如果性能成为了问题（也请不要事到临头才意识到它），更高效地处理我们的服务器是件好事。那么，我们如何更高效地管理服务器端呢？我们可以维护一个进入的连接池，一定数量的ConnectionHandler 将为它提供服务。这种设计能带来以下好处：

  • 它限定了允许同时连接的数目。
  • 我们只需启动 ConnectionHandler Thread 一次。

幸运的是，跟在我们的多线程示例中一样，往代码中添加“池”不需要来一个大改动。事实上，应用程序的客户机端根本就不受影响。在服务器端，我们在服务器启动时创建一定数量的ConnectionHandler，我们把进入的连接放入“池”中并让 ConnectionHandler 打理剩下的事情。这种设计中有很多我们不打算讨论的可能存在的技巧。例如，我们可以通过限定允许在“池”中建立的连接的数目来拒绝客户机。

请注意：我们将不会再次讨论 acceptConnections()。这个方法跟前面示例中的完全一样。它无限循环地调用 ServerSocket 上的 accept() 并把连接传递到 handleConnection()。

 2. 创建 PooledRemoteFileServer 类

这里是 PooledRemoteFileServer 类的结构：

import java.io.*;
import java.net.*;
import java.util.*;

public class PooledRemoteFileServer {
    protected int maxConnections;
    protected int listenPort;
    protected ServerSocket serverSocket;

    public PooledRemoteFileServer(int aListenPort, int maxConnections) {
        listenPort = aListenPort;
        this.maxConnections = maxConnections;
    }
    public static void main(String[] args) {
    }
    public void setUpHandlers() {
    }
    public void acceptConnections() {
    }
    protected void handleConnection(Socket incomingConnection) {
    }
}


 

请注意一下您现在应该熟悉了的 import 语句。我们给类以下实例变量以保存：

  • 我们的服务器能同时处理的活动客户机连接的最大数目

  • 进入的连接的侦听端口（我们没有指定缺省值，但如果您想这样做，并不会受到限制）

  • 将接受客户机连接请求的 ServerSocket

类的构造器用的参数是侦听端口和连接的最大数目

我们的类有一个 main() 方法和三个其它方法。稍后我们将探究这些方法的细节。现在只须知道setUpHandlers() 创建数目为maxConnections 的大量PooledConnectionHandler，而其它两个方法则与我们前面已经看到的相似：acceptConnections() 在ServerSocket 上侦听传入的客户机连接，而handleConnection 则在客户机连接一旦被建立后就实际处理它。


 3. 实现 main()

这里我们实现需作改动的 main() 方法，该方法将创建能够处理给定数目的客户机连接的 PooledRemoteFileServer，并告诉它接受连接：

public static void main(String[] args) {
    PooledRemoteFileServer server = new PooledRemoteFileServer(3000, 3);
    server.setUpHandlers();
    server.acceptConnections();
}

我们的 main() 方法很简单。我们实例化一个新的PooledRemoteFileServer，它将通过调用setUpHandlers() 来建立三个PooledConnectionHandler。一旦服务器就绪，我们就告诉它acceptConnections()。

 4. 建立连接处理程序

 

public void setUpHandlers() {
    for (int i = 0; i < maxConnections; i++) {
        PooledConnectionHandler currentHandler = new PooledConnectionHandler();
        new Thread(currentHandler, "Handler " + i).start();
    }
}


 

setUpHandlers() 方法创建 maxConnections（例如 3）个PooledConnectionHandler 并在新 Thread 中激活它们。用实现了 Runnable 的对象来创建Thread 使我们可以在 Thread 调用 start() 并且可以期望在 Runnable 上调用了 run()。换句话说，我们的 PooledConnectionHandler 将等着处理进入的连接，每个都在它自己的Thread 中进行。我们在示例中只创建三个 Thread，而且一旦服务器运行，这就不能被改变。


 5. 处理连接

这里我们实现需作改动的 handleConnections() 方法，它将委派 PooledConnectionHandler 处理连接：

protected void handleConnection(Socket connectionToHandle) {
PooledConnectionHandler.processRequest(connectionToHandle);
}


 

我们现在叫 PooledConnectionHandler 处理所有进入的连接（processRequest() 是一个静态方法）。

这里是 PooledConnectionHandler 类的结构：

import java.io.*;
import java.net.*;
import java.util.*;

public class PooledConnectionHandler implements Runnable {
    protected Socket connection;
    protected static List pool = new LinkedList();

    public PooledConnectionHandler() {
    }
    public void handleConnection() {
    }
    public static void processRequest(Socket requestToHandle) {
    }
    public void run() {
    }
}


 

这个助手类与 ConnectionHandler 非常相似，但它带有处理连接池的手段。该类有两个实例变量：

  • connection 是当前正在处理的Socket
  • 名为 pool 的静态LinkedList 保存需被处理的连接


 6. 填充连接池

这里我们实现 PooledConnectionHandler 上的 processRequest() 方法，它将把传入请求添加到池中，并告诉其它正在等待的对象该池已经有一些内容：

public static void processRequest(Socket requestToHandle) {
    synchronized (pool) {
        pool.add(pool.size(), requestToHandle);
        pool.notifyAll();
    }
}


 

理解这个方法要求有一点关于 Java 的关键字 synchronized 如何工作的背景知识。我们将简要讲述一下线程。

先来看一些定义：

  • 原子方法。在执行过程中不能被中断的方法（或代码块）
  • 互斥锁。客户机欲执行原子方法时必须获得的单个“锁”

因此，当对象 A 想使用对象 B 的 synchronized 方法doSomething() 时，对象 A 必须首先尝试获取对象 B 的互斥锁。是的，这意味着当对象 A 拥有该互斥锁时，没有其它对象可以调用对象 B 上任何其它synchronized 方法。

synchronized 块是个稍微有些不同的东西。您可以同步任何对象上的一个块，而不只是在本身的某个方法中含有该块的对象。在我们的示例中，processRequest() 方法包含有一个pool（请记住它是一个LinkedList，保存等待处理的连接池）的synchronized块。我们这样做的原因是确保没有别人能跟我们同时修改连接池。

既然我们已经保证了我们是唯一“涉水”池中的人，我们就可以把传入的 Socket 添加到LinkedList 的尾端。一旦我们添加了新的连接，我们就用以下代码通知其它正在等待该池的Thread，池现在已经可用：


pool.notifyAll();

Object 的所有子类都继承这个notifyAll() 方法。这个方法，连同我们下一屏将要讨论的wait() 方法一起，就使一个Thread 能够让另一个Thread 知道一些条件已经具备。这意味着该第二个Thread 一定正在等待那些条件的满足。


 7. 从池中获取连接

这里我们实现 PooledConnectionHandler 上需作改动的 run()方法，它将在连接池上等待，并且池中一有连接就处理它：

public void run() {
        while (true) {
             synchronized (pool) {
                  while (pool.isEmpty()) {
                       try {
                            pool.wait();
                       } catch (InterruptedException e) {
                            return;
                       }
                   }
                   connection = (Socket) pool.remove(0);
             }
             handleConnection();
        }
}


 

回想一下在前一屏讲过的：一个 Thread 正在等待有人通知它连接池方面的条件已经满足了。在我们的示例中，请记住我们有三个PooledConnectionHandler 在等待使用池中的连接。每个PooledConnectionHandler 都在它自已的Thread 中运行，并通过调用pool.wait() 产生阻塞。当我们的processRequest() 在连接池上调用notifyAll() 时，所有正在等待的PooledConnectionHandler 都将得到“池已经可用”的通知。然后各自继续前行调用pool.wait()，并重新检查while(pool.isEmpty()) 循环条件。除了一个处理程序，其它池对所有处理程序都将是空的，因此，在调用pool.wait() 时，除了一个处理程序，其它所有处理程序都将再次产生阻塞。恰巧碰上非空池的处理程序将跳出while(pool.isEmpty())
循环并攫取池中的第一个连接：


connection = (Socket) pool.remove(0);

处理程序一旦有一个连接可以使用，就调用 handleConnection() 处理它。

在我们的示例中，池中可能永远不会有多个连接，只是因为事情很快就被处理掉了。如果池中有一个以上连接，那么其它处理程序将不必等待新的连接被添加到池。当它们检查
pool.isEmpty() 条件时，将发现其值为假，然后就从池中攫取一个连接并处理它。

还有另一件事需注意。当 run() 拥有池的互斥锁时，processRequest() 如何能够把连接放到池中呢？答案是对池上的wait() 的调用释放锁，而wait() 接着就在自己返回之前再次攫取该锁。这就使得池对象的其它同步代码可以获取该锁。

8. 处理连接：再一次

这里我们实现需做改动的 handleConnection() 方法，该方法将攫取连接的流，使用它们，并在任务完成之后清除它们：

public void handleConnection() {
    try {
        PrintWriter streamWriter = new PrintWriter(connection.getOutputStream());
        BufferedReader streamReader =
            new BufferedReader(new InputStreamReader(connection.getInputStream()));

        String fileToRead = streamReader.readLine();
        BufferedReader fileReader = new BufferedReader(new FileReader(fileToRead));

        String line = null;
        while ((line = fileReader.readLine()) != null)
            streamWriter.println(line);

        fileReader.close();
        streamWriter.close();
        streamReader.close();
    } catch (FileNotFoundException e) {
        System.out.println("Could not find requested file on the server.");
    } catch (IOException e) {
        System.out.println("Error handling a client: " + e);
    }
}


 

跟在多线程服务器中不同，我们的 PooledConnectionHandler 有一个handleConnection() 方法。这个方法的代码跟非池式的ConnectionHandler 上的run() 方法的代码完全一样。首先，我们把OutputStream 和InputStream 分别包装进（用Socket 上的 getOutputStream() 和getInputStream()）BufferedReader 和PrintWriter。然后我们逐行读目标文件，就象我们在多线程示例中做的那样。再一次，我们获取一些字节之后就把它们放到本地的line 变量中，然后写出到客户机。完成读写操作之后，我们关闭FileReader 和打开的流。


9. 总结一下带有连接池的服务器

我们的带有连接池的服务器研究完了。让我们回顾一下创建和使用“池版”服务器的步骤：

  1. 创建一个新种类的连接处理程序（我们称之为 PooledConnectionHandler）来处理池中的连接。

  2. 修改服务器以创建和使用一组 PooledConnectionHandler。

   附： PooledRemoteFileServer 的完整代码清单

import java.io.*;
import java.net.*;
import java.util.*;

public class PooledRemoteFileServer {
    protected int maxConnections;
    protected int listenPort;
    protected ServerSocket serverSocket;
    public PooledRemoteFileServer(int aListenPort, int maxConnections) {
        listenPort = aListenPort;
        this.maxConnections = maxConnections;
    }
    public void acceptConnections() {
        try {
            ServerSocket server = new ServerSocket(listenPort, 5);
            Socket incomingConnection = null;
            while (true) {
                incomingConnection = server.accept();
                handleConnection(incomingConnection);
            }
        } catch (BindException e) {
            System.out.println("Unable to bind to port " + listenPort);
        } catch (IOException e) {
            System.out.println("Unable to instantiate a ServerSocket on port: " + listenPort);
        }
    }
    protected void handleConnection(Socket connectionToHandle) {
        PooledConnectionHandler.processRequest(connectionToHandle);
    }
    public static void main(String[] args) {
        PooledRemoteFileServer server = new PooledRemoteFileServer(3000, 3);
        server.setUpHandlers();
        server.acceptConnections();
    }
    public void setUpHandlers() {
        for (int i = 0; i < maxConnections; i++) {
            PooledConnectionHandler currentHandler = new PooledConnectionHandler();
            new Thread(currentHandler, "Handler " + i).start();
        }
    }
}


PooledConnectionHandler 的完整代码清单

 

import java.io.*;
import java.net.*;
import java.util.*;

public class PooledConnectionHandler implements Runnable {
    protected Socket connection;
    protected static List pool = new LinkedList();
    public PooledConnectionHandler() {
    }
    public void handleConnection() {
        try {
            PrintWriter streamWriter = new PrintWriter(connection.getOutputStream());
            BufferedReader streamReader = new BufferedReader(new InputStreamReader(connection.getInputStream()));

            String fileToRead = streamReader.readLine();
            BufferedReader fileReader = new BufferedReader(new FileReader(fileToRead));

            String line = null;
            while ((line = fileReader.readLine()) != null)
                streamWriter.println(line);

            fileReader.close();
            streamWriter.close();
            streamReader.close();
        } catch (FileNotFoundException e) {
            System.out.println("Could not find requested file on the server.");
        } catch (IOException e) {
            System.out.println("Error handling a client: " + e);
        }
    }
    public static void processRequest(Socket requestToHandle) {
        synchronized (pool) {
            pool.add(pool.size(), requestToHandle);
            pool.notifyAll();
        }
    }
    public void run() {
        while (true) {
            synchronized (pool) {
                while (pool.isEmpty()) {
                    try {
                        pool.wait();
                    } catch (InterruptedException e) {
                        return;
                    }
                }
                connection = (Socket) pool.remove(0);
            }
            handleConnection();
        }
    }
}


 

标签：

相关文章