Socket 多线程演进与线程池使用

Dear 丶 2023-05-31 12:19 9阅读 0赞

**目录**

一个简单的线程示例

如何将Socket客户端连接植入线程？

通过初始化线程个数模拟一个线程池

使用Java Executor ExecutorService线程池管理Socket多线程

--------------------

##  一个简单的线程示例 ##

实现一个线程可以继承Thread类或者实现Runnable接口：

package Chapter4;
    
    import java.util.concurrent.TimeUnit;
    /**
     * @TODO     (功能说明:线程例子)
     * @author   PJL
     * @package  Chapter4
     * @motto    学习需要毅力，那就秀毅力
     * @file     ThreadExample.java
     * @time     2019年10月24日 下午10:11:58
     */
    public class ThreadExample implements Runnable {
    
    	private String greeting; // Message to print to console
    
    	public ThreadExample(String greeting) {
    		this.greeting = greeting;
    	}
    
    	public void run() {
    		while (true) {
    			System.out.println(Thread.currentThread().getName() + ":  " + greeting);
    			try {
    				// Sleep 0 to 100 milliseconds
    				TimeUnit.MILLISECONDS.sleep(((long) Math.random() * 100));
    			} catch (InterruptedException e) {
    			} // Should not happen
    		}
    	}
    
    	public static void main(String[] args) {
    		new Thread(new ThreadExample("Hello")).start();
    		new Thread(new ThreadExample("Aloha")).start();
    		new Thread(new ThreadExample("Ciao")).start();
    	}
    }

## 如何将Socket客户端连接植入线程？ ##

当我们在服务端serverSocket.accept()接收到一个新的连接时，我们获取到的实际上是一个客户端连接的Socket实例。通过这个实例，我们可以实现数据的双向通信。将Socket作为参数放入线程，是为了让Socket流水作业变成线程式工作方式。

package Chapter4;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.OutputStream;
    import java.net.Socket;
    import java.util.logging.Level;
    import java.util.logging.Logger;
    /**
     * 
     * @TODO     (功能说明:Socket日志处理线程)
     * @author   PJL
     * @package  Chapter4
     * @motto    学习需要毅力，那就秀毅力
     * @file     EchoProtocol.java
     * @time     2019年10月24日 下午10:20:18
     */
    public class EchoProtocol implements Runnable {
    	private static final int BUFSIZE = 32; // Size (in bytes) of I/O buffer
    	private Socket clntSock; // Socket connect to client
    	private Logger logger; // Server logger
    
    	public EchoProtocol(Socket clntSock, Logger logger) {
    		this.clntSock = clntSock;
    		this.logger = logger;
    	}
    
    	/**
    	 * Socket处理日志方法
    	 * @param clntSock
    	 * @param logger
    	 */
    	public static void handleEchoClient(Socket clntSock, Logger logger) {
    		try {
    			// Get the input and output I/O streams from socket
    			InputStream in = clntSock.getInputStream();
    			OutputStream out = clntSock.getOutputStream();
    
    			int recvMsgSize; // Size of received message
    			int totalBytesEchoed = 0; // Bytes received from client
    			byte[] echoBuffer = new byte[BUFSIZE]; // Receive Buffer
    			// Receive until client closes connection, indicated by -1
    			while ((recvMsgSize = in.read(echoBuffer)) != -1) {
    				out.write(echoBuffer, 0, recvMsgSize);
    				totalBytesEchoed += recvMsgSize;
    			}
    
    			logger.info("Client " + clntSock.getRemoteSocketAddress() + ", echoed " + totalBytesEchoed + " bytes.");
    
    		} catch (IOException ex) {
    			logger.log(Level.WARNING, "Exception in echo protocol", ex);
    		} finally {
    			try {
    				clntSock.close();
    			} catch (IOException e) {
    			}
    		}
    	}
    
    	/**
    	 * 线程执行方法
    	 */
    	public void run() {
    		handleEchoClient(clntSock, logger);
    	}
    }

接下来就是将线程利用起来了，请看下面：

package Chapter4;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    import java.util.logging.Logger;
    /**
     * @TODO     (功能说明:一客户一线程方式处理Socket)
     * @author   PJL
     * @package  Chapter4
     * @motto    学习需要毅力，那就秀毅力
     * @file     TCPEchoServerThread.java
     * @time     2019年10月24日 下午10:29:46
     */
    public class TCPEchoServerThread {
    
    	public static void main(String[] args) throws IOException {
    		
    		if (args.length == 0) {
    			args = new String[1];
    			args[0] = "7402";
    		}
    
    		if (args.length != 1) { // Test for correct # of args
    			throw new IllegalArgumentException("Parameter(s): <Port>");
    		}
    
    		int echoServPort = Integer.parseInt(args[0]); // Server port
    
    		// Create a server socket to accept client connection requests
    		ServerSocket servSock = new ServerSocket(echoServPort);
    
    		Logger logger = Logger.getLogger("practical");
    
    		// Run forever, accepting and spawning a thread for each connection
    		while (true) {
    			Socket clntSock = servSock.accept(); // Block waiting for connection
    			// Spawn thread to handle new connection
    			Thread thread = new Thread(new EchoProtocol(clntSock, logger));
    			thread.start();
    			logger.info("Created and started Thread " + thread.getName());
    		}
    		/* NOT REACHED */
    	}
    }

## 通过初始化线程个数模拟一个线程池 ##

线程池的作用就是为了让线程能够被管理起来合理地使用，合理利用服务器资源。如果一个用户一个线程，那么我们的资源随着用户的增长而愈来愈紧张，所以为了避免线程的滥用，线程池的使用很重要。

package Chapter4;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    import java.util.logging.Level;
    import java.util.logging.Logger;
    /**
     * @TODO     (功能说明:模拟线程池等待Socket请求处理)
     * @author   PJL
     * @package  Chapter4
     * @motto    学习需要毅力，那就秀毅力
     * @file     TCPEchoServerPool.java
     * @time     2019年10月24日 下午10:36:47
     */
    public class TCPEchoServerPool {
    
    	public static void main(String[] args) throws IOException {
    		
    		if (args.length == 0) {
    			args = new String[2];
    			args[0] = "7402";
    			args[1] = "10";
    		}
    
    		if (args.length != 2) { // Test for correct # of args
    			throw new IllegalArgumentException("Parameter(s): <Port> <Threads>");
    		}
    
    		int echoServPort = Integer.parseInt(args[0]); // Server port
    		int threadPoolSize = Integer.parseInt(args[1]);
    
    		// Create a server socket to accept client connection requests
    		final ServerSocket servSock = new ServerSocket(echoServPort);
    
    		final Logger logger = Logger.getLogger("practical");
    
    		// Spawn a fixed number of threads to service clients
    		for (int i = 0; i < threadPoolSize; i++) {
    			Thread thread = new Thread() {
    				public void run() {
    					while (true) {
    						try {
    							Socket clntSock = servSock.accept(); // Wait for a connection
    							EchoProtocol.handleEchoClient(clntSock, logger); // Handle it
    						} catch (IOException ex) {
    							logger.log(Level.WARNING, "Client accept failed", ex);
    						}
    					}
    				}
    			};
    			thread.start();
    			logger.info("Created and started Thread = " + thread.getName());
    		}
    	}
    }

上面初始化了一定数量的线程，每个线程在没有收到请求时都是阻塞的，我们不知道哪一个线程会幸运地连接到新的请求，不确定性就是线程存在最有意思的地方。

## 使用Java Executor ExecutorService线程池管理Socket多线程 ##

Executor 是线程池的基础接口类。

Executors 是线程池的一组实现方法，如下面：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jvb255YXhubg_size_16_color_FFFFFF_t_70][]

我们可以这样来使用线程池对象：

int threadPoolSize=100;
    Executor service0 =Executors.newFixedThreadPool(threadPoolSize);
    
    Executor service1 = Executors.newCachedThreadPool();
    
    ExecutorService service2=Executors.newCachedThreadPool();

Executor的方法很少：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jvb255YXhubg_size_16_color_FFFFFF_t_70 1][]

ExecutorService提供了更为高级的方法：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jvb255YXhubg_size_16_color_FFFFFF_t_70 2][]

下面来看一个Socket关于线程池的使用方式：

package Chapter4;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    import java.util.concurrent.Executor;
    import java.util.concurrent.Executors;
    import java.util.logging.Logger;
    /**
     * @TODO     (功能说明:Java自带高级线程池管理Socket请求线程服务)
     * @author   PJL
     * @package  Chapter4
     * @motto    学习需要毅力，那就秀毅力
     * @file     TCPEchoServerExecutor.java
     * @time     2019年10月24日 下午10:45:45
     */
    public class TCPEchoServerExecutor {
    
    	public static void main(String[] args) throws IOException {
    		
    		if (args.length == 0) {
    			args = new String[1];
    			args[0] = "7402";
    		}
    
    		if (args.length != 1) { // Test for correct # of args
    			throw new IllegalArgumentException("Parameter(s): <Port>");
    		}
    
    		int echoServPort = Integer.parseInt(args[0]); // Server port
    
    		// Create a server socket to accept client connection requests
    		ServerSocket servSock = new ServerSocket(echoServPort);
    
    		Logger logger = Logger.getLogger("practical");
    
    		/*
    		 * int threadPoolSize=100; 
    		 * Executor service =Executors.newFixedThreadPool(threadPoolSize);
    		 */
    		Executor service = Executors.newCachedThreadPool(); // Dispatch svc
    
    		// Run forever, accepting and spawning threads to service each connection
    		while (true) {
    			Socket clntSock = servSock.accept(); // Block waiting for connection
    			service.execute(new EchoProtocol(clntSock, logger));
    			// set timeout socket thread
    			//service.execute(new TimelimitEchoProtocol(clntSock, logger));
    		}
    		/* NOT REACHED */
    	}
    }

另外Socket连接还比较关注阻塞问题，所以会涉及到超时问题的处理，主要是对Socket设置超时时间，如下例：

package Chapter4;
    
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.OutputStream;
    import java.net.Socket;
    import java.util.logging.Level;
    import java.util.logging.Logger;
    /**
     * 
     * @TODO     (功能说明:时间限制-控制接收receive、读取read、等待链接accept 超时时间 核心是设置setSoTimeout)
     * @author   PJL
     * @package  Chapter4
     * @motto    学习需要毅力，那就秀毅力
     * @file     TimeLimitEchoProtocol.java
     * @time     2019年10月24日 下午10:54:00
     */
    class TimelimitEchoProtocol implements Runnable {
    	private static final int BUFSIZE = 32; // Size (bytes) buffer
    	private static final String TIMELIMIT = "10000"; // Default limit (ms)
    	private static final String TIMELIMITPROP = "Timelimit"; // Thread property
    
    	private static int timelimit;
    	private Socket clntSock;
    	private Logger logger;
    
    	public TimelimitEchoProtocol(Socket clntSock, Logger logger) {
    		this.clntSock = clntSock;
    		this.logger = logger;
    		// Get the time limit from the System properties or take the default
    		timelimit = Integer.parseInt(System.getProperty(TIMELIMITPROP, TIMELIMIT));
    	}
    
    	public static void handleEchoClient(Socket clntSock, Logger logger) {
    
    		try {
    			// Get the input and output I/O streams from socket
    			InputStream in = clntSock.getInputStream();
    			OutputStream out = clntSock.getOutputStream();
    			int recvMsgSize; // Size of received message
    			int totalBytesEchoed = 0; // Bytes received from client
    			byte[] echoBuffer = new byte[BUFSIZE]; // Receive buffer
    			long endTime = System.currentTimeMillis() + timelimit;
    			int timeBoundMillis = timelimit;
    
    			clntSock.setSoTimeout(timeBoundMillis);
    			// Receive until client closes connection, indicated by -1
    			while ((timeBoundMillis > 0) && // catch zero values
    					((recvMsgSize = in.read(echoBuffer)) != -1)) {
    				out.write(echoBuffer, 0, recvMsgSize);
    				totalBytesEchoed += recvMsgSize;
    				timeBoundMillis = (int) (endTime - System.currentTimeMillis());
    				clntSock.setSoTimeout(timeBoundMillis);
    			}
    			logger.info("Client " + clntSock.getRemoteSocketAddress() + ", echoed " + totalBytesEchoed + " bytes.");
    		} catch (IOException ex) {
    			logger.log(Level.WARNING, "Exception in echo protocol", ex);
    		}
    	}
    
    	public void run() {
    		handleEchoClient(this.clntSock, this.logger);
    	}
    }

学习代码实例请参看：[https://github.com/Nuclear-Core-Learning/TCPIP-Socket/tree/master/src/Chapter4][https_github.com_Nuclear-Core-Learning_TCPIP-Socket_tree_master_src_Chapter4]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jvb255YXhubg_size_16_color_FFFFFF_t_70]: /images/20230531/0db7ab60934549459040fb5c1f829cf4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jvb255YXhubg_size_16_color_FFFFFF_t_70 1]: /images/20230531/f9610e0e37264297a48fa555330d2bef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jvb255YXhubg_size_16_color_FFFFFF_t_70 2]: /images/20230531/47cc74bd032c432a8cb22c3b4b90922c.png
[https_github.com_Nuclear-Core-Learning_TCPIP-Socket_tree_master_src_Chapter4]: https://github.com/Nuclear-Core-Learning/TCPIP-Socket/tree/master/src/Chapter4