线程阻塞

线程阻塞队列

文章分类:Java编程

摘自别人，仅供分析

/*
编写一个线程安全、大小固定的队列
提供阻塞式的方法put，若队列没有空间，则方法put会一直等待
提供阻塞式的方法take，若队列为空，则方法take会一直等待
启动30个线程操作该队列，每个线程进行一次put和一次take操作
*/

/*
已经按照顺序获得读锁和写锁了，但是如果启动30个线程的话，基本上每次都会死锁，线程都停在read_lock.wait()的位置，
如果启动20个线程就只有一半的几率会死锁（其实都在等待read_lock的锁，不能说是死锁），但每一个线程take一次必然会put一次，
或者反过来，按说是不会有都等待read_lock的情况
*/

package com.huawei.test;

import java.util.*;

public class Queue1
{
   final int SIZE = 10; //队列固定大小
   ArrayList store = new ArrayList(SIZE);

   Object write_lock = new Object();//用于对store的写操作，如get/add/set/remove
   Object read_lock = new Object(); //用于对store只读操作，如取size

   public Queue1(){}

   public void put (Object o) //没有空间一直等待
   {
       while(true){
           synchronized(read_lock){
               try{
                   if(store.size() == SIZE){
                       read_lock.wait();//如果队列已满，就释放锁
                   }else{
                       synchronized(write_lock){
                           Thread.sleep(50);
                           store.add(o); //增加元素到队列
                           System.out.println(Thread.currentThread().getName() + "****PUT::Size=" + store.size());
                           Thread.sleep(50);
                           read_lock.notifyAll(); //通知其他线程
                           break;
                         }
                   }
               }catch(Exception ex){
                       ex.printStackTrace(System.err);
               }
           }
       }
   }


   public Object take () //没有数据一直等待
   {
       while(true){
           synchronized(read_lock){
               try{
                   if(store.size() == 0){
                       read_lock.wait();//如果队列没有数据，就释放锁
                   }else{
                       synchronized(write_lock){
                           Thread.sleep(50);
                           Object obj = store.remove(0); //从队列头移走数据
                           System.out.println(Thread.currentThread().getName() + "****Take::Size=" + store.size());
                           Thread.sleep(50);
                           read_lock.notifyAll();//通知其他线程
                           return obj;
                       }
                   }
               }catch(Exception ex){
                   ex.printStackTrace(System.err);
               }
           }
       }
   }


   public static void main(String[] args){
       Queue1 queue1 = new Queue1(); //创建一个队列

       for(int i = 0; i < 30; i++){ //启动30个线程访问队列
           TestThread thread = new TestThread(queue1,i);
           System.out.println( "--Thread:" + i + " Start!" );
           thread.start();
           try{
               Thread.sleep(10); //没隔十毫秒启动一个线程
           }catch(Exception ex){
               ex.printStackTrace(System.err);
           }
       }
   }

}


class TestThread extends Thread
{
   Queue1 queue1 = null;
   int sn = 0;

   public TestThread(Queue1 queue1,int sn){
       this.queue1 = queue1;
       this.sn = sn;
       setName("Thread::" + sn); //以序号作为线程名
   }

   public void run(){
       String tmp = null;
       try{
           if( sn < 7){ //sn小于7的线程先put，后take
               tmp = "Thread-PUT::" + sn + "---put::";
               queue1.put(tmp);
               Thread.sleep(10);
               tmp = "Thread-Take::" + sn + "---take::";
               Object obj = queue1.take();
           }else{ //sn大于7的线程先take，后put
               tmp = "Thread-Take::" + sn + "---take::";
               Object obj = queue1.take();
               Thread.sleep(10);
               tmp = "Thread-PUT::" + sn + "---put::";
               queue1.put(tmp);
           }
           System.out.println("Thread::" + sn + " task over!");
       }catch(Exception ex){
           ex.printStackTrace(System.err);
       }

   }
}