伪异步IO编程：服务端使用线程池实现(三)

客户端

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

/**
 * 客户端
 */
public class Client {
    public static void main(String[] args) {
        try {
            // 1、请求与服务端的Socket对象链接
            Socket socket = new Socket("127.0.0.1",9999);
            // 2、得到一个打印流
            PrintStream ps = new PrintStream(socket.getOutputStream());
            // 3、使用循环不断地发送消息给服务端接收
            Scanner sc = new Scanner(System.in);
            while (true){
                System.out.print("请说：");
                String msg = sc.nextLine();
                ps.println(msg);
                ps.flush();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

服务端

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

/**
 * 服务端：开发实现伪异步通信架构
 */
public class Server {
    public static void main(String[] args) {
        try {
            // 1、注册端口
            ServerSocket ss = new ServerSocket(9999);
            // 2、定义一个循环接收客户端的Socket链接请求
            // 初始化一个线程池对象
            HandlerSocketServerPool pool = new HandlerSocketServerPool(6,10);
            while (true){
                Socket socket = ss.accept();
                // 3、把socket对象交给一个线程池进行处理
                // 把socket封装成一个任务对象交给线程池处理
                Runnable target = new ServerRunnableTarget(socket);
                pool.execute(target);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}

服务端线程池

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 ThreadPoolExecutor(int corePoolSize,
     *                               int maximumPoolSize,
     *                               long keepAliveTime,
     *                               TimeUnit unit,
     *                               BlockingQueue<Runnable> workQueue)
     */
    public HandlerSocketServerPool(int maxThreadNum , int queueSize){
        executorService = new ThreadPoolExecutor(3,maxThreadNum,120, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(queueSize));
    }

    /**
     * 3、提供一个方法来提交任务给线程池的任务队列来暂存，等着线程池来处理
     */
    public void execute(Runnable target){
        executorService.execute(target);
    }
}

服务端线程

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 socket){
        this.socket = socket;
    }
    @Override
    public void run(){
        // 处理接收到的客户端Socket通信需求
        InputStream is = null;
        try {
            // 从socket对象中得到一个字节输入流
            is = socket.getInputStream();
        } catch (IOException e) {
            e.printStackTrace();
        }
        // 使用缓冲字符输入流包装字节输入流
        BufferedReader br = new BufferedReader(new InputStreamReader(is));
        String msg = null;
        while (true){
            try {
                if (!(( msg = br.readLine() ) != null)) {
                    break;
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println(msg);
        }
    }
}