Java语言深入 多线程程序模型研究

共有 6 条评论

1. balini 于 2009-07-22 11:41:34发表:

   很深奥噢...

2. tjt1127 于 2009-07-21 10:26:37发表:

   顶了

3. zhqs 于 2006-10-30 10:53:34发表:

   学习雪鞋!

4. fengmayi1 于 2006-10-29 00:05:09发表:

   class Scanner implements Runnable{
   String host;
   int port;
   
   Scanner(String host, int port){
   this.host = host;
   this.port = port;
   }
   
   public void run(){
   Socket s = null;
   try{
   s = new Socket(InetAddress.getByName(host),port);
   System.out.println("The port of " + port + " is opened.");
   }catch(IOException ex){
   }finally{
   try{
   if(s != null) s.close();
   }catch(IOException e){
   }
   }
   }
   }
   ---------------------------------------------------------------------------------------------------------------------
   PortScannerByThreadPool是主程序类，处理输入的4个参数（和第3个程序是一样的）：主机名、开始端口、结束端口和线程数。Scanner类定义了真正的”任务“。在PortScannerByThreadPool中通过new ThreadPool(nThreads)创建ThreadPool对象，然后在for循环中通过new Scanner(host,i)创建”任务“对象，再通过tp.execute(ps)把”任务“对象添加到”池“中。
   
   读者可以编译运行第4个程序，得出的结果和前面的是一样的。但是第4和第3个程序之间最大的差别就是：第4个程序会一直运行下去，不会自动结束。在第3个程序中存在一个isFinished()方法，可以用来判断任务是否处理完毕，而第4个程序中没有这样做。请读者自己思考这个问题。
   
   在第3和第4个程序中，我们可以概括出多线程的模型。第3个程序的线程”池“里装的要处理的对象，第4个程序的线程”池“里装的是”工人“，还需要通过定义”任务“并给把它”派工“给”工人“。我个人比较偏好后者的线程池模型，虽然类的个数多了几个，但逻辑很清晰。不管怎样，第3和第4个程序中关键的部分都大同小异，就是2个synchronized程序块中的内容，如下（第4个程序中的）：
   synchronized(queue) {
   queue.addLast(r);
   queue.notifyAll();
   }
   和
   synchronized(queue) {
   while (queue.isEmpty()) {
   try{
   queue.wait();
   }catch (InterruptedException ignored){
   }
   }
   
   r = (Runnable) queue.removeFirst();
   }
   
   一般拿synchronized用来定义方法或程序块，这样可以在多线程同时访问的情况下，保证在一个时刻只能有一个线程对这部分内容进行访问，避免了数据出错。在第3个程序中通过List entries = Collections.synchronizedList(new LinkedList())来定义”池“，在第4个程序中直接用LinkedList queue，都差不多，只是Collections.synchronizedList()可以保证”池“的同步，其实”池“里的内容访问都是在synchronized定义的程序块中，所以不用Collections.synchronizedList()也是可以的。
   
   wait()和notifyAll()是很重要的，而且这2个方法是Object基类的方法，所以任何一个类都是可以使用的。这里说明一个可能产生混淆的问题：queue.wait()并不是说queue对象需要进行等待，而是说queue.wait()所在的线程需要进行等待，并且释放对queue的锁，把对queue的访问权交给别的线程。如果读者对这2个方法难以理解，建议参考JDK的文档说明。
   
   好了，通过以上4个例子的理解，读者应该能对多线程的程序设计有了一定的理解。第3和第4个程序对应线程模型是非常重要的，可以说是多线程程序设计过程中不可或缺的内容。
   
   如果读者对以上的内容有任何疑问，可以和我联系，qianh@cntmi.com 版权所有，严禁转载
   
   参考资料：
   1、《Java Networking Programming, 3rd》written by Elliotte Rusty Harold, Published by O'Reilly,2004
   2、“Thread pools and work queues” written by Brian Goetz, Principal Consultant, Quiotix Corp.

5. fengmayi1 于 2006-10-29 00:04:46发表:

   以上程序需要4个参数，输入java PortScanner 10.1.1.182 1 10000 100运行（第4个参数是线程数），结果前两个程序一样，但是速度比第一个要快，可能比第二个要慢一些。
   
   第3个程序是把端口作为“池”中的对象，下面我们看第4个实现方式，把“池”里面的对象定义成是线程类，把具体的任务定义成”池“中线程类的参数。第4个程序有2个文件组成，分别是ThreadPool.java和PortScannerByThreadPool.java.
   ThreadPool.java文件内容如下：
   -----------------------------------------------------------
   import java.util.LinkedList;
   
   public class ThreadPool{
   private final int nThreads;
   private final PoolWorker[] threads;
   private final LinkedList queue;
   
   public ThreadPool(int nThreads){
   this.nThreads = nThreads;
   queue = new LinkedList();
   threads = new PoolWorker[nThreads];
   
   for (int i=0; i threads[i] = new PoolWorker();
   threads[i].start();
   }
   }
   
   public void execute(Runnable r) {
   synchronized(queue) {
   queue.addLast(r);
   queue.notifyAll();
   }
   }
   
   private class PoolWorker extends Thread {
   public void run() {
   Runnable r;
   
   while (true) {
   synchronized(queue) {
   while (queue.isEmpty()) {
   try{
   queue.wait();
   }catch (InterruptedException ignored){
   }
   }
   
   r = (Runnable) queue.removeFirst();
   }
   
   try {
   r.run();
   }
   catch (RuntimeException e) {
   }
   }
   }
   }
   }
   ------------------------------------------------------------------------------------------------------------------
   在ThreadPool.java文件中定义了2个类：ThreadPool和PoolWorker。ThreadPool类中的nThreads变量表示线程数，PoolWorker数组类型的threads变量表示线程池中的“工人”，这些“工人”的工作就是一直循环处理通过queue.addLast(r)加入到“池”中的任务。
   PortScannerByThreadPool.java文件内容如下：
   -------------------------------------------------------------------------------------------------------------------
   import java.io.IOException;
   import java.net.InetAddress;
   import java.net.Socket;
   
   public class PortScannerByThreadPool {
   public static void main(String[] args) {
   String host = null;
   int beginport = 1;
   int endport = 65535;
   int nThreads = 100;
   try{
   host = args[0];
   beginport = Integer.parseInt(args[1]);
   endport = Integer.parseInt(args[2]);
   nThreads = Integer.parseInt(args[3]);
   if(beginport <= 0 || endport >= 65536 || beginport > endport){
   throw new Exception("Port is illegal");
   }
   }catch(Exception e){
   System.out.println("Usage: java PortScannerSingleThread host beginport endport nThreads");
   System.exit(0);
   }
   
   ThreadPool tp = new ThreadPool(nThreads);
   
   for(int i = beginport; i <= endport; i++){
   Scanner ps = new Scanner(host,i);
   tp.execute(ps);
   }
   }
   }

6. fengmayi1 于 2006-10-29 00:04:23发表:

   以上程序在for循环结构中创建PortProcessor对象，PortProcessor类是线程类，其关键的Socket在public void run()方法中实现。此程序比第一个单线程的程序运行效率提高很多倍，几乎在几秒钟内得出结果。所以可见多线程处理是何等的重要。
   程序(java PortScannerMultiThread 10.1.1.100 1 1000)运行结果如下：
   The port 25 is opened at 10.1.1.100
   The port 42 is opened at 10.1.1.100
   The port 88 is opened at 10.1.1.100
   ...
   
   仔细对第2个程序分析，不难发现其中的问题：创建的线程个数是不固定的，取决于输入的第二和第三个参数。如果扫描1～100端口，那么主线程就产生100个线程来分别处理；如果扫描1~10000端口，主线程就会产生10000个线程来进行处理。在JVM中创建如此多的线程同样会带来性能上的问题，因为线程的创建和消失都是需要花费系统资源的。所以以上的第二个程序也存在明显的不足。
   
   所以，我们需要一个确定数量的线程在JVM中运行，这样就需要了解“线程池”(ThreadPool)的概念。线程池在多线程程序设计中是比不可少的，而且初学者不太容易掌握，下面通过对线程池的介绍，结合第3和第4个程序，引出两种常用的线程池模型。
   
   第一种实现线程池的方法是：创建一个”池“，在”池“中增加要处理的数据对象，然后创建一定数量的线程，这些线程对”池“中的对象进行处理。当”池“是空的时候，每个线程处于等待状态；当往”池“里添加一个对象，通知所有等待的线程来处理（当然一个对象只能有一个线程来处理）。
   
   第二种方法是：同样创建一个”池“，但是在”池“中放的不是数据对象，而是线程，可以把”池“中的一个个线程比喻成一个个”工人“，当没有任务的时候，”工人“们严阵以待；当给”池“添加一个任务后，”工人“就开始处理并直到处理完成。
   
   在第3个程序中，定义了List类型的entries作为“池”，这个“池”用来保存需要扫描的端口，List中的元素必须是Object类型，不能用基本数据类型int往池里添加，而需要用使用Integer。在processMethod()方法中，首先就启动一定数量的PortThread线程，同时在while循环中通过entries.add(0, new Integer(port))往“池”里添加对象。在PortThread类的run()方法中通过entry = (Integer)entries.remove(entries.size()-1);取得“池”中的对象，转换成int后传递给Socket构造方法。
   第3个程序如下：
   -----------------------------------------------------------------------------------------------------------------------
   import java.io.IOException;
   import java.net.InetAddress;
   import java.net.Socket;
   import java.net.UnknownHostException;
   import java.util.Collections;
   import java.util.LinkedList;
   import java.util.List;
   
   public class PortScanner {
   private List entries = Collections.synchronizedList(new LinkedList()); //这个”池“比较特别
   int numofthreads;
   static int port;
   int beginport;
   int endport;
   InetAddress remote = null;
   
   public boolean isFinished(){
   if(port >= endport){
   return true;
   }else{
   return false;
   }
   }
   
   PortScanner(InetAddress addr, int beginport, int endport, int numofthreads){
   this.remote = addr;
   this.beginport = beginport;
   this.endport = endport;
   this.numofthreads = numofthreads;
   }
   
   public void processMethod(){
   for(int i = 0; i < numofthreads; i++){ //创建一定数量的线程并运行
   Thread t = new PortThread(remote, entries, this);
   t.start();
   }
   
   port = beginport;
   
   while(true){
   if(entries.size() > numofthreads){
   try{
   Thread.sleep(1000); //”池“中的内容太多的话就sleep
   }catch(InterruptedException ex){
   
   }
   continue;
   }
   
   synchronized(entries){
   if(port > endport) break;
   entries.add(0, new Integer(port)); //往”池“里添加对象，需要使用int对应的Integer类
   entries.notifyAll();
   port++;
   }
   }
   }
   
   public static void main(String[] args) {
   String host = null;
   int beginport = 1;
   int endport = 65535;
   int nThreads = 100;
   try{
   host = args[0];
   beginport = Integer.parseInt(args[1]);
   endport = Integer.parseInt(args[2]);
   nThreads = Integer.parseInt(args[3]);
   if(beginport <= 0 || endport >= 65536 || beginport > endport){
   throw new Exception("Port is illegal");
   }
   }catch(Exception e){
   System.out.println("Usage: java PortScannerSingleThread host beginport endport nThreads");
   System.exit(0);
   }
   
   try{
   PortScanner scanner = new PortScanner(InetAddress.getByName(host), beginport, endport, nThreads);
   scanner.processMethod();
   }catch(UnknownHostException ex){
   }
   }
   }
   
   class PortThread extends Thread{
   private InetAddress remote;
   private List entries;
   PortScanner scanner;
   
   PortThread(InetAddress add, List entries, PortScanner scanner){
   this.remote = add;
   this.entries = entries;
   this.scanner = scanner;
   }
   
   public void run(){
   Integer entry;
   while(true){
   synchronized(entries){
   while(entries.size() == 0){
   if(scanner.isFinished()) return;
   try{
   entries.wait(); //”池“里没内容就只能等了
   }catch(InterruptedException ex){
   }
   }
   entry = (Integer)entries.remove(entries.size()-1); //把”池“里的东西拿出来进行处理
   }
   
   Socket s = null;
   
   try{
   s = new Socket(remote, entry.intValue());
   System.out.println("The port of " + entry.toString() + " of the remote " + remote +" is opened.");
   
   }catch(IOException e){
   }finally{
   try{
   if(s != null) s.close();
   }catch(IOException e){
   
   }
   }
   }
   }
   }