IO之我见（2）伪异步IO编程

上一篇讲了BIO通信，服务端连接一个客户端就要创建一个线程，这样会有很多缺点，具体请看上一篇文章 https://blog.csdn.net/Chen_leilei/article/details/123461331

这一篇我们采用线程池和任务队列实现，当客户端接入时，将客户端的Socket封装成一个Task(实现Runnable接口交给后端的线程池进行处理),jdk线程池维护一个消息队列和N个活跃的线程，对消息队列中的Socket任务进行处理，由于线程池可以设置消息队列的大小和最大线程数，因此，他的资源占用是可控的，无论多少个客户端并发访问，都不会造成宕机。

Server代码

package com.chen.four;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;


public class Server {
    public static void main(String[] args) {
        //1.注册端口
        try {
            ServerSocket serverSocket = new ServerSocket(9999);
            //初始化一个线程池对象
            HandlerSocketServerPool pool = new HandlerSocketServerPool(3,100);
            //2.定义一个循环接收客户端的请求
            while (true){
                Socket accept = serverSocket.accept();
                //3把socket对象包装成一个任务对象交给线程池

                pool.execute(new ServerRunnableTarget(accept));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

 Client代码

package com.chen.four;

import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;
import java.net.Socket;
import java.util.Scanner;

public class Client {
    public static void main(String[] args) throws IOException {
        try {
            Socket socket = new Socket("127.0.0.7",9999);
            OutputStream outputStream = socket.getOutputStream();
            PrintStream ps = new PrintStream(outputStream);
            Scanner scanner = new Scanner(System.in);
            while (true){
                System.out.print("请说");
                String s = scanner.nextLine();
                ps.println(s);
                ps.flush();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

线程池

package com.chen.four;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class HandlerSocketServerPool {
    //1.创建一个线程池的成员变量，用于存储一个线程池对象
    private ExecutorService executorService;
    //2.创建这个类的对象的时候，就需要初始化线程池对象
    public HandlerSocketServerPool(int maxThreadNum, int queueSize) {
        executorService =new ThreadPoolExecutor(
                3,
                maxThreadNum,
                120L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue(queueSize));
    }
    //3 提供一个方法来提交任务给线程池的任务队列来暂存，等着线程池来处理

    public void  execute(Runnable runnable){
        executorService.execute(runnable);
    }
}

       实际线程代码

package com.chen.four;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.Socket;

public class ServerRunnableTarget implements Runnable{
    private Socket socket;
    public ServerRunnableTarget(Socket accept) {
        this.socket = accept;
    }

    @Override
    public void run() {

        InputStream is = null;
        try {
            is = socket.getInputStream();
            BufferedReader bf = new BufferedReader(new InputStreamReader(is));
            String msg;
            while ((msg= bf.readLine())!=null){
                System.out.println("服务端收到消息"+msg);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

小结：伪异步IO采用了线程池实现，因此避免了为每个请求创建一个独立线程造成线程资源消耗尽的问题，但由于底层依然采用的是同步阻塞模型，无法从根本上解决问题

如果单个消息处理的缓慢，或者服务端线程池中的线程全部都正在被使用，那么后续的socket消息都会在队列中排队，新的Socket请求将会被拒绝，客户端会发生大量的连接超时