java 网络编程（2.3）-----------采用线程池多线程的Socket 通信

 吃了个泡面，看了会盘龙，然后放了一首歌《17岁那年的雨季》（随机的 - -）

 

好了 。接着来说多线程吧，为什么把两骗分开来呢，可能是这篇比较难的

 

两个类

EchoServer.java

 

package leaningSocket;
import java.io.*;
import java.net.*;
public class EchoServer {
private int port=6500;
private ServerSocket serverSocket;
private ThreadPool threadPool;  //线程池
private final int POOL_SIZE=4;  //单个CPU时线程池中工作线程的数目

public EchoServer() throws IOException {
serverSocket = new ServerSocket(port);
//创建线程池
//Runtime的availableProcessors()方法返回当前系统的CPU的数目
//系统的CPU越多，线程池中工作线程的数目也越多
threadPool= new ThreadPool(
Runtime.getRuntime().availableProcessors() * POOL_SIZE);

System.out.println("服务器启动");
}

public void service() {
while (true) {
Socket socket=null;
try {
socket = serverSocket.accept();
threadPool.execute(new Handler(socket)); //把与客户通信的任务交给线程池
}catch (IOException e) {
e.printStackTrace();
}
}
}

public static void main(String args[])throws IOException {
new EchoServer().service();
}
}

class Handler implements Runnable{
private Socket socket;
public Handler(Socket socket){
this.socket=socket;
}
private PrintWriter getWriter(Socket socket)throws IOException{
OutputStream socketOut = socket.getOutputStream();
return new PrintWriter(socketOut,true);
}
private BufferedReader getReader(Socket socket)throws IOException{
InputStream socketIn = socket.getInputStream();
return new BufferedReader(new InputStreamReader(socketIn));
}
public String echo(String msg) {
return "echo:" + msg;
}
public void run(){
try {
System.out.println("New connection accepted " +
socket.getInetAddress() + ":" +socket.getPort());
BufferedReader br =getReader(socket);
PrintWriter pw = getWriter(socket);

String msg = null;
while ((msg = br.readLine()) != null) {
System.out.println(msg);
pw.println(echo(msg));
if (msg.equals("bye"))
break;
}
}catch (IOException e) {
e.printStackTrace();
}finally {
try{
if(socket!=null)socket.close();
}catch (IOException e) {e.printStackTrace();}
}
}
}

/****************************************************
* 作者：孙卫琴                                     *
* 来源：<<Java网络编程精解>>                       *
* 技术支持网址：www.javathinker.org                *
***************************************************/

 

好吧，这个代码跟前面的基本上没什么区别，除了那句

threadPool.execute(new Handler(socket)); //把与客户通信的任务交给线程池

 

就是说为每个客户链接分配一个线程不同，当然咯，我们采用线程池了嘛。

 

ThreadPool.java

package leaningSocket;

import java.util.LinkedList;

public class ThreadPool extends ThreadGroup
{

private static int threadPoolID = 0 ;/**表示线程池ID*/
private LinkedList<Runnable> workQueue;/**表示工作队列**/
private boolean isClosed = false; /**线程池是否关闭*/
private int threadID ; /**表示工作线程ID */
/**线程池构造函数*/
public ThreadPool( int poolSize) {

super("ThreadPool-"+( threadPoolID++));

System.out.println("当前线程池大小为"+poolSize);
// TODO Auto-generated constructor stub
setDaemon(true);

workQueue = new LinkedList<Runnable>();

System.out.println("我要启动所有线程咯");

for(int i = 0 ; i < poolSize ; i ++)
new WorkThread().start();
}
/**向工作队列中加入一个新任务,有工作线程去执行该任务*/
public synchronized void execute( Runnable task)
{
if ( isClosed)//线程池被关则抛出IllegalStateException异常
{
System.out.println("当前线程池关闭状态");
throw new  IllegalStateException();
}
if ( task !=null)
{
workQueue.add(task);
System.out.println("已经将一个任务房到工作队列中了");
System.out.println("......正在唤醒一个线程，给我去工作");
notify();//唤醒正在getTask()方法中等待任务的工作线程
}
}
/**从工作队列中取出一个任务，工作线程会调用此方法*/
protected synchronized Runnable getTask() throws InterruptedException
{
while( workQueue.size()==0)
{
if(isClosed) return null;
System.out.println("当前队列中没有任务，我睡觉了");

/**如果把wait（）去掉就会一直输出。。我要睡觉了：（*/
wait();//如果工作队列中没有任务，就等待任务
}
/**移除并返回此列表的第一个元素。*/
return workQueue.removeFirst();
}
/** 关闭线程池 */
public synchronized void close()
{
if(!isClosed)
{
isClosed = true;
workQueue.clear();//清空工作队列
interrupt(); //中断所有的工作线程，该方法继承自ThreadGroup类
}
}

/** 等待工作线程把所有任务执行完 */
public void join ()
{
synchronized( this)
{
isClosed = true;
notifyAll();//唤醒还在getTask()方法中等待任务的工作线程
}

Thread[] threads = new Thread[this.activeCount()];
//enumerate()方法继承自ThreadGroup类，获得线程组中当前所有活着的工作线程
int count = enumerate(threads);
for (int i=0; i<count; i++) { //等待所有工作线程运行结束
try {
threads[i].join();  //等待工作线程运行结束
}catch(InterruptedException ex) { }
}
}

private class WorkThread extends Thread
{
public WorkThread()
{
//加入到当前ThreadPool线程组中
super(ThreadPool.this,"WorkThread-" + (threadID++));
System.out.println("当前调用线程"+(threadID-1));
}
public void run() {
while (!this.isInterrupted()) {  //isInterrupted()方法继承自Thread类，判断线程是否被中断
Runnable task = null;
try { //得到任务
task = getTask();
}catch (InterruptedException ex){}

// 如果getTask()返回null或者线程执行getTask()时被中断，则结束此线程
if (task == null) return;

try { //运行任务，捕获异常
task.run();
} catch (Throwable t) {
t.printStackTrace();
}
}//#while
}//#run()
}//#WorkThread类
}

 

哎，这个类就难多了，看了哥好久，对多线程还是没能把握呀。。

 

 

那么只好说下2个小时里面我干的事了：

 

首先运行EchoServer.java

结果：

当前线程池大小为4
我要启动所有线程咯
当前调用线程0
当前调用线程1
当前调用线程2
当前调用线程3
服务器启动
当前队列中没有任务，我睡觉了
当前队列中没有任务，我睡觉了
当前队列中没有任务，我睡觉了
当前队列中没有任务，我睡觉了

这个输出说明什么呢,说明了线程池的执行顺序，其实他是这么执行的首先传入poolSize大小，自然就会调用ThreadPool.java的构造函数了，然后构造函数创建相应的线程个数，而后，每个线程都会去getTask()...

这个是什么呢？就是到任务队列去寻找有没有任务，这里任务队列是LinkedList<Runnable> workQueue

 

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

好。。分下段
---------------------------------------------------------------------------------------------------------------------

这里要题下，因为是线程，所以就会不段运行（我理解为死循环可以么？），那么实际上在getTask()函数里

有个wait(),我一开始不明白，后来我把他去掉后发现他会不断输出“我要睡觉了”，而其实本来是只输出4个

我要睡觉了，因为只有四个线程麽。。。。

 

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

好了，一开始代码的运行顺序我知道了，然后我就要测试他是怎么执行任务的了，于是打开cmd,

输入telnet localhost 6500

 结果：

已经将一个任务房到工作队列中了
......正在唤醒一个线程，给我去工作
New connection accepted /127.0.0.1:3438

 

观看代码里的输出信息就可以发现，当客户端连接时，首先是EchServer.java

执行代码 threadPool.execute(new Handler(socket)); 

 

然后在 ThreadPool.java中把该任务放到workQueue中，再唤醒一个线程（我不知道是不是随即唤醒的）

 

被唤醒的线程就会继续检查workQueue中是否有任务，这个时候自然会发现有任务咯。。然后就会把任务队列中

 

第一个任务移除。。。

 

 

值得一题的是（其实不值一提） 线程池中的线程数目是4个，我无聊得用5个客户端去访问6500端口，然后用第5个客户端

发送消息--------------结果自然是没有反应咯。。等到关闭第一个客户端，消息框中就会输出刚才发送的内容。。也说明了

任务队列是有序的（好像是P话 - -）。。

 

 

好了。。这个就是我们后来2个多小时的结果。。。阿门

 

然后记下还没看懂的（其实还有很多没看懂）：

 

疑问：

 

首先super()其实并不是很懂了。。只是简单理解为调用父类构造函数，哎，上课没怎么听，不过这次主要是为了理解

网络编程，应该关系不大，以后再看吧，

 

然后在ThreadPool里的一个setDaemon(true);也不清楚啥意思，算了。丢掉

 

当然还会发现，在ThreadPool里的join()与close()要这么调用呢？？？

 

然后最最最不懂的就是synchronized了，贴段百度解释吧：

 

synchronized 方法控制对类成员变量的访问：每个类实例对应一把锁，每个 synchronized 方法都必须获得调用该方法的类实例的锁方能执行，否则所属线程阻塞，方法一旦执行，就独占该锁，直到从该方法返回时才将锁释放，此后被阻塞的线程方能获得该锁，重新进入可执行状态。这种机制确保了同一时刻对于每一个类实例，其所有声明为 synchronized 的成员函数中至多只有一个处于可执行状态（因为至多只有一个能够获得该类实例对应的锁），从而有效避免了类成员变量的访问冲突（只要所有可能访问类成员变量的方法均被声明为 synchronized）。

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

 

下面是摘自网上对ThreadPool.java的理解：

 

 

 在ThreadPool类中定义了一个LinkedList类型的workQueue成员变量，它表示工作队列，用来存放线程池要执行的任务，每个 任务都是Runnable实例。ThreadPool类的客户程序（利用ThreadPool来执行任务的程序）只要调用ThreadPool类的 execute (Runnable task)方法，就能向线程池提交任务。在ThreadPool类的execute()方法中，先判断线程池是否已 经关闭。如果线程池已经关闭，就不再接收任务，否则就把任务加入到工作队列中，并且唤醒正在等待任务的工作线程。

 

在ThreadPool类的构造方法中，会创建并启动若干工作线程，工作线程的数目由构造方法的参数 poolSize决定。WorkThread类表示工作线程，它是ThreadPool类的内部类。工作线程从工作队列中取出一个任务，接着执行该任务， 然后再从工作队列中取出下一个任务并执行它，如此反复。

工作线程从工作队列中取任务的操作是由ThreadPool类的getTask()方法实现的，它的处理逻辑如下：

◆如果队列为空并且线程池已关闭，那就返回null，表示已经没有任务可以执行了；

◆如果队列为空并且线程池没有关闭，那就在此等待，直到其他线程将其唤醒或者中断；

◆如果队列中有任务，就取出第一个任务并将其返回。

线程池的join()和close()方法都可用来关闭线程池。join()方法确保在关闭线程池之前，工作线程把队列中的所有任务都执行完。而close()方法则立即清空队列，并且中断所有的工作线程。

ThreadPool类是ThreadGroup类的子类。ThreadGroup类表示线程组，它提供了一些管理线程组中线程的方法。例如， interrupt()方法相当于调用线程组中所有活着的线程的interrupt()方法。线程池中的所有工作线程都加入到当前ThreadPool对 象表示的线程组中。

ThreadPool类在close()方法中调用了interrupt()方以上interrupt()方法用于中断所有的工作线程。interrupt()方法会对工作线程造成以下影响：

◆如果此时一个工作线程正在ThreadPool的getTask()方法中因为执行wait()方法而阻塞，则会抛出InterruptedException；

◆如果此时一个工作线程正在执行一个任务，并且这个任务不会被阻塞，那么这个工作线程会正常执行完任务，但是在执行下一轮while (!isInterrupted()) {…}循环时，由于isInterrupted()方法返回true，因此退出while循环。

 

 

 

 3.6.4  使用线程池的注意事项(其实我还没看过^^)

虽然线程池能大大提高服务器的并发性能，但使用它也会存在一定风险。与所有多线程应用程序一样，用线程池构建的应用程序容易产生各种并发问题，如对 共享资源的竞争和死锁。此外，如果线程池本身的实现不健壮，或者没有合理地使用线程池，还容易导致与线程池有关的死锁、系统资源不足和线程泄漏等问题。

1．死锁

任何多线程应用程序都有死锁风险。造成死锁的最简单的情形是，线程A持有对象X的锁，并且在等待对象Y的锁，而线程B持有对象Y的锁，并且在等待对象X的锁。线程A与线程B都不释放自己持有的锁，并且等待对方的锁，这就导致两个线程永远等待下去，死锁就这样产生了。

虽然任何多线程程序都有死锁的风险，但线程池还会导致另外一种死锁。在这种情形下，假定线程池中的所有工作线程都在执行各自任务时被阻塞，它们都在 等待某个任务A的执行结果。而任务A依然在工作队列中，由于没有空闲线程，使得任务A一直不能被执行。这使得线程池中的所有工作线程都永远阻塞下去，死锁 就这样产生了。

2．系统资源不足

如果线程池中的线程数目非常多，这些线程会消耗包括内存和其他系统资源在内的大量资源，从而严重影响系统性能。

3．并发错误

线程池的工作队列依靠wait()和notify()方法来使工作线程及时取得任务，但这两个方法都难于使用。

如果编码不正确，可能会丢失通知，导致工作线程一直保持空闲状态，无视工作队列中需要处理的任务。因此使用这些方法时，必须格外小心，即便是专家也 可能在这方面出错。最好使用现有的、比较成熟的线程池。例如，直接使用java.util.concurrent包中的线程池类。

4．线程泄漏

使用线程池的一个严重风险是线程泄漏。对于工作线程数目固定的线程池，如果工作线程在执行任务时抛出RuntimeException 或Error，并 且这些异常或错误没有被捕获，那么这个工作线程就会异常终止，使得线程池永久失去了一个工作线程。如果所有的工作线程都异常终止，线程池就最终变为空，没 有任何可用的工作线程来处理任务。

导致线程泄漏的另一种情形是，工作线程在执行一个任务时被阻塞，如等待用户的输入数据，但是由于用户一直不输入数据（可能是因为用户走开了），导致 这个工作线程一直被阻塞。这样的工作线程名存实亡，它实际上不执行任何任务了。假如线程池中所有的工作线程都处于这样的阻塞状态，那么线程池就无法处理新 加入的任务了。

5．任务过载

当工作队列中有大量排队等候执行的任务时，这些任务本身可能会消耗太多的系统资源而引起系统资源缺乏。

综上所述，线程池可能会带来种种风险，为了尽可能避免它们，使用线程池时需要遵循以下原则。

（1）如果任务A在执行过程中需要同步等待任务B的执行结果，那么任务A不适合加入到线程池的工作队列中。如果把像任务A一样的需要等待其他任务执行结果的任务加入到工作队列中，可能会导致线程池的死锁。

（2）如果执行某个任务时可能会阻塞，并且是长时间的阻塞，则应该设定超时时间，避免工作线程永久的阻塞下去而导致线程泄漏。在服务器程序中，当线程等待客户连接，或者等待客户发送的数据时，都可能会阻塞。可以通过以下方式设定超时时间：

◆调用ServerSocket的setSoTimeout(int timeout)方法，设定等待客户连接的超时时间，参见本章3.5.1节（SO_TIMEOUT选项）；

◆对于每个与客户连接的Socket，调用该Socket的setSoTimeout(int timeout)方法，设定等待客户发送数据的超时时间，参见本书第2章的2.5.3节（SO_TIMEOUT选项）。

（3）了解任务的特点，分析任务是执行经常会阻塞的I/O操作，还是执行一直不会阻塞的运算操作。前者时断时续地占用CPU，而后者对CPU具有更高的利用率。预计完成任务大概需要多长时间？是短时间任务还是长时间任务？

根据任务的特点，对任务进行分类，然后把不同类型的任务分别加入到不同线程池的工作队列中，这样可以根据任务的特点，分别调整每个线程池。

（4）调整线程池的大小。线程池的最佳大小主要取决于系统的可用CPU的数目，以及工作队列中任务的特点。假如在一个具有 N 个CPU的系统上只 有一个工作队列，并且其中全部是运算性质（不会阻塞）的任务，那么当线程池具有 N 或 N+1 个工作线程时，一般会获得最大的 CPU 利用率。

如果工作队列中包含会执行I/O操作并常常阻塞的任务，则要让线程池的大小超过可用CPU的数目，因为并不是所有工作线程都一直在工作。选择一个典 型的任务，然后估计在执行这个任务的过程中，等待时间（WT）与实际占用CPU进行运算的时间（ST）之间的比例WT/ST。对于一个具有N个CPU的系 统，需要设置大约N×(1+WT/ST)个线程来保证CPU得到充分利用。

当然，CPU利用率不是调整线程池大小过程中唯一要考虑的事项。随着线程池中工作线程数目的增长，还会碰到内存或者其他系统资源的限制，如套接字、打开的文件句柄或数据库连接数目等。要保证多线程消耗的系统资源在系统的承载范围之内。

（5）避免任务过载。服务器应根据系统的承载能力，限制客户并发连接的数目。当客户并发连接的数目超过了限制值，服务器可以拒绝连接请求，并友好地告知客户：服务器正忙，请稍后再试。

文章出处：http://www.diybl.com/course/3_program/java/javaxl/2008229/102026_5.html