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

线程演进与线程池使用"/>

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

目录

一个简单的线程示例

如何将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 consolepublic ThreadExample(String greeting) {this.greeting = greeting;}public void run() {while (true) {System.out.println(Thread.currentThread().getName() + ":  " + greeting);try {// Sleep 0 to 100 millisecondsTimeUnit.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.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 bufferprivate Socket clntSock; // Socket connect to clientprivate Logger logger; // Server loggerpublic 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 socketInputStream in = clntSock.getInputStream();OutputStream out = clntSock.getOutputStream();int recvMsgSize; // Size of received messageint totalBytesEchoed = 0; // Bytes received from clientbyte[] echoBuffer = new byte[BUFSIZE]; // Receive Buffer// Receive until client closes connection, indicated by -1while ((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.ServerSocket;
import java.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 argsthrow new IllegalArgumentException("Parameter(s): <Port>");}int echoServPort = Integer.parseInt(args[0]); // Server port// Create a server socket to accept client connection requestsServerSocket servSock = new ServerSocket(echoServPort);Logger logger = Logger.getLogger("practical");// Run forever, accepting and spawning a thread for each connectionwhile (true) {Socket clntSock = servSock.accept(); // Block waiting for connection// Spawn thread to handle new connectionThread 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.ServerSocket;
import java.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 argsthrow new IllegalArgumentException("Parameter(s): <Port> <Threads>");}int echoServPort = Integer.parseInt(args[0]); // Server portint threadPoolSize = Integer.parseInt(args[1]);// Create a server socket to accept client connection requestsfinal ServerSocket servSock = new ServerSocket(echoServPort);final Logger logger = Logger.getLogger("practical");// Spawn a fixed number of threads to service clientsfor (int i = 0; i < threadPoolSize; i++) {Thread thread = new Thread() {public void run() {while (true) {try {Socket clntSock = servSock.accept(); // Wait for a connectionEchoProtocol.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 是线程池的一组实现方法，如下面：

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

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

Executor的方法很少：

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

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

package Chapter4;import java.io.IOException;
import java.ServerSocket;
import java.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 argsthrow new IllegalArgumentException("Parameter(s): <Port>");}int echoServPort = Integer.parseInt(args[0]); // Server port// Create a server socket to accept client connection requestsServerSocket 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 connectionwhile (true) {Socket clntSock = servSock.accept(); // Block waiting for connectionservice.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.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) bufferprivate static final String TIMELIMIT = "10000"; // Default limit (ms)private static final String TIMELIMITPROP = "Timelimit"; // Thread propertyprivate 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 defaulttimelimit = Integer.parseInt(System.getProperty(TIMELIMITPROP, TIMELIMIT));}public static void handleEchoClient(Socket clntSock, Logger logger) {try {// Get the input and output I/O streams from socketInputStream in = clntSock.getInputStream();OutputStream out = clntSock.getOutputStream();int recvMsgSize; // Size of received messageint totalBytesEchoed = 0; // Bytes received from clientbyte[] echoBuffer = new byte[BUFSIZE]; // Receive bufferlong endTime = System.currentTimeMillis() + timelimit;int timeBoundMillis = timelimit;clntSock.setSoTimeout(timeBoundMillis);// Receive until client closes connection, indicated by -1while ((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);}
}

学习代码实例请参看：