线程池
线程池介绍
在创建一个线程时存在一定的开销,创建线程的开销相比于一般的算法来说较大。首先需要建立一个调用栈,接着操作系统还需要建立很多数据结构来维护线程的状态等等。为了解决每次都需要临时创建线程带来开销的问题,引入了线程池。
线程池会预先建立好线程,等待任务派发。一个线程池内可能会有几十上百个闲置的线程,当有任务来临时,需要选中一个线程执行任务。执行完毕后释放该线程资源,重新回到闲置状态,等待下一个任务。
在线程池中,通常有一个Blocking Queue用来派发任务。队列通常为空,当有新的任务进队列时,闲置线程开始抢占该任务。当没有闲置线程时,新任务到达Blocking Queue便会开始排队,等待有线程空出来。
线程池的参数
corePoolSize:线程池中初始线程的数量,可能处于等待状态
maimumPoolSize:线程池中最大允许线程数量
keepAliveTime:超出corePoolSize部分线程如果等待这些时间,就会被回收
Java线程池ExecutorTester
可以使用ThreadPoolExecutor创建线程池:
参数包含了corePoolSize,maimumPoolSize,keepAliveTime,时间单元TimeUnit以及一个可执行的workQueue任务队列。
另一种写法是用executors.newFixedThreadPool()创建线程池。这样的写法参数较少,只需要指定一个线程数即可。
ExecutorService executor = Executors.newFixedThreadPool(3);
如此创建的线程池,拥有的线程数量固定为3,如果任务数大于3,那么多余的任务则只有排队等待。
接下来我们一口气给这个线程池派发10个任务试试:
for (int i = 0; i < 10; i++) {
executor.submit(new CodingTask(i));
}
System.out.println("10 tasks dispatched successfully.");
由于线程池参数为3,所以会按3个一组来抢占任务。
第一轮派发完毕后:
可以看到将3个任务派发给了线程们,workQueue中还剩7个任务。
Future<?>可以用来控制任务的执行:
cancel(boolean):可以临时中止执行任务
get():若成功执行则返回null,否则等待
iscancelled():获取任务是否中止
isDone():获取任务是否成功执行
ExecutorService executor = Executors.newFixedThreadPool(3);
List<Future<?>> taskResults = new LinkedList<>();
for (int i = 0; i < 10; i++) {
taskResults.add(executor.submit(new CodingTask(i)));
}
System.out.println("10 tasks dispatched successfully.");
for (Future<?> taskResult : taskResults) {
taskResult.get();//get等待task结束,一旦结束会返回null
}
服务器socket编程
普通socket编程
我们设想这样一个场景,一个服务器监听在6666端口上。若客户发aaa,则响应Hello aaa。首先我们需要建立一个服务端socket去监听端口号:
ServerSocket serverSocket = new ServerSocket(6666)
然后建立一个客户端socket,用getRemoteSocketAddress()方法获取远端socket端口号:
Socket clientSocket = serverSocket.accept();
System.out.println("Incoming connection from "
+ clientSocket.getRemoteSocketAddress());
最后用clientSoclet提供的getInputStream()方法和getOutputStream()方法实现开头我们所设想的功能:
try(ServerSocket serverSocket = new ServerSocket(6666)){
System.out.println("listening on "+serverSocket.getLocalSocketAddress());
Socket clientSocket = serverSocket.accept();
System.out.println("Incoming connection from "+clientSocket.getRemoteSocketAddress());
try(Scanner input = new Scanner(clientSocket.getInputStream())){
String request = input.nextLine();
System.out.println(String.format("Request from %s: %s",
clientSocket.getRemoteSocketAddress(),
request));
String response = "Hello"+request+".\n";
clientSocket.getOutputStream().write(response.getBytes());
}
}
执行以上代码并在命令行执行:
telnet localhost 6666
输入aaa后,服务端返回消息:
当然,正常的服务器不会在处理完一个请求后就马上切断与客户端的连接。所以,我们需要在把处理请求块放进一个循环中,并且加入终止条件即可。
try(ServerSocket serverSocket = new ServerSocket(6666)){
System.out.println("listening on "+serverSocket.getLocalSocketAddress());
Socket clientSocket = serverSocket.accept();
System.out.println("Incoming connection from "+clientSocket.getRemoteSocketAddress());
try(Scanner input = new Scanner(clientSocket.getInputStream())){
while(true) {
String request = input.nextLine();
if(request.equals("quit")) {
break;
}
System.out.println(String.format("Request from %s: %s",
clientSocket.getRemoteSocketAddress(),
request));
String response = "Hello "+request+".\n";
clientSocket.getOutputStream().write(response.getBytes());
}
}
}
但是此时的服务器只能处理一个客户的请求。当另开一个控制台telnet 6666端口时,由于已经被占用,其他客户无法接受服务。
线程池并行处理客户请求
线程池在前面已经介绍过,它预先建立好了线程,等待任务来临。在线程池中有一个Blocking Queue,任务到来后会进入该队列,当到达队头并且有闲置进程时就会被选中执行。
在Java中,扔进Blocking Queue的是一个个Client Handler,因为每个client会占据一个线程,所以Client Handler即代表待执行的任务。线程调用Client Handle的run()方法来执行该任务。
ExecutorService executor = Executors.newFixedThreadPool(3);
RequestHandler requestHandler = new RequestHandler();
try (ServerSocket serverSocket = new ServerSocket(7777)) {
System.out.println("Listening on "
+ serverSocket.getLocalSocketAddress());
while (true) {
Socket clientSocket = serverSocket.accept();
System.out.println("Incoming connection from "
+ clientSocket.getRemoteSocketAddress());
executor.submit(
new ClientHandler(clientSocket, requestHandler));
}
}
此时,服务器端就就可以支持多线程工作,因为线程池最大为3,所以最多可以同时开3个命令行访问服务器。当第4个命令行请求服务时,仍可连接成功,但需等待前3个服务中至少有一个服务退出,才可以接受服务。
Java NIO异步处理客户请求
每个线程处理一个客户请求看似十分合理,实则也存在缺点。因为线程池的容量有限,一个几十上百线程的线程池最多也只能服务几十上百个客户,难以实现大容量高吞吐的服务端。
在上一节的服务端线程池中,看似是所有线程都在忙着应付手头上的任务,而无暇顾及新到的任务,然而事实真的如此吗?比如当线程等待用户的输入时,用户的输入可以非常慢,这时线程便会被阻塞,并非在忙碌工作着。
因此,我们是否可以将所有client的input记住,再去检查谁输入完毕,就先去读这个输入并处理。也就是说,我们应该把一个个的request去交给线程处理,而不是把整个client都交给线程去处理。这样就避免了整个线程的等待时间,即将同步处理优化为了异步处理,这就是NIO异步服务器。
文章来源:https://www.toymoban.com/news/detail-831154.html
在NIO异步服务器中,等待的任务不再进入一个Blocking Queue,而是进入一个Channel列表。列表中有很多的Client,在每个循环开始时,通过一个Selector选择器去选择一个有数据的Channel,接着Request Handle去处理这个有数据的Channel,如此循环。文章来源地址https://www.toymoban.com/news/detail-831154.html
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.bind(new InetSocketAddress(8888));
System.out.println("Listening on "
+ serverChannel.getLocalAddress());
Selector selector = Selector.open();
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
ByteBuffer buffer = ByteBuffer.allocate(1024);
RequestHandler requestHandler = new RequestHandler();
while (true) {
int selected = selector.select();
if (selected == 0) {
continue;
}
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIter = selectedKeys.iterator();
while (keyIter.hasNext()) {
SelectionKey key = keyIter.next();
if (key.isAcceptable()) {
ServerSocketChannel channel =
(ServerSocketChannel) key.channel();
SocketChannel clientChannel = channel.accept();
System.out.println("Incoming connection from "
+ clientChannel.getRemoteAddress());
clientChannel.configureBlocking(false);
clientChannel.register(
selector, SelectionKey.OP_READ);
}
if (key.isReadable()) {
SocketChannel channel =
(SocketChannel) key.channel();
channel.read(buffer);
String request = new String(buffer.array()).trim();
buffer.clear();
System.out.println(String.format(
"Request from %s: %s",
channel.getRemoteAddress(),
request));
String response = requestHandler.handle(request);
channel.write(ByteBuffer.wrap(response.getBytes()));
}
keyIter.remove();
}
}
到了这里,关于Java学习小记——多线程&Socket编程的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!