调度模式·Worker-Channel-Request

public   class  Channel
{
     private   static   final   int  MAX_REQUEST  =   100 ;

     private   final  IRequest[] requestQueue;

     private   int  tail;  //  下一个putRequest的地方

     private   int  head;  //  下一个takeRequest的地方

     private   int  count;  //  Request的数量
    
     private   boolean  stoped  =   true ;

     private   final  WorkerThread[] threadPool;

     public  Channel( int  threads)
    {
         this .requestQueue  =   new  Request[MAX_REQUEST];
         this .head  =   0 ;
         this .tail  =   0 ;
         this .count  =   0 ;

        threadPool  =   new  WorkerThread[threads];
         for  ( int  i  =   0 ; i  <  threadPool.length; i ++ )
        {
            threadPool[i]  =   new  WorkerThread( this );
        }
    }

     public   void  startWorkers()
    {
         this .stoped  =   false ;
         for  ( int  i  =   0 ; i  <  threadPool.length; i ++ )
        {
            threadPool[i].start();
        }
    }
    
     public   synchronized   int  getCount()
    {
         return   this .count;
    }
    
     public   synchronized   boolean  isStoped()
    {
         return   this .stoped;
    }
    
     public   synchronized   void  stopWorkers()
    {
         this .stoped  =   true ;
    }

     public   synchronized   void  putRequest(IRequest request)
    {
         while  (count  >=  requestQueue.length)
        {
             try
            {
                wait();
            }
             catch  (InterruptedException e)
            {
            }
        }
        requestQueue[tail]  =  request;
        tail  =  (tail  +   1 )  %  requestQueue.length;
        count ++ ;
        notifyAll();
    }

     public   synchronized  IRequest takeRequest()
    {
         while  (count  <=   0 )
        {
             try
            {
                wait();
            }
             catch  (InterruptedException e)
            {
            }
        }
        IRequest request  =  requestQueue[head];
        head  =  (head  +   1 )  %  requestQueue.length;
        count -- ;
        notifyAll();
         return  request;
    }
}

public   interface  IRequest
{
     public   abstract   void  execute()  throws  Exception;
}

public   class  WorkerThread  extends  Thread
{
     private   final  Channel channel;

     public  WorkerThread(Channel channel)
    {
         this .channel  =  channel;
    }

     public   void  run()
    {
        System.out.println( " [Thread]: WorkerThread  "   +   this .getName()
                 +   "  start! " );

         while  ( true )
        {
             if  (channel.getCount()  <=   0   &&  channel.isStoped())
            {
                System.out.println( " [Thread]: WorkerThread  "   +   this .getName()
                         +   "  stop! " );
                stop();
            }
             else
            {
                IRequest request  =  channel.takeRequest();
                 try
                {
                    request.execute();
                }
                 catch  (Exception e)
                {
                    System.out.println(e.getMessage());
                }
            }
        }
    }
}