java并发中锁的应用_JAVA_编程开发_程序员俱乐部

中国优秀的程序员网站程序员频道CXYCLUB技术地图
热搜:
更多>>
 
您所在的位置: 程序员俱乐部 > 编程开发 > JAVA > java并发中锁的应用

java并发中锁的应用

 2014/11/14 22:06:41  hacksin  程序员俱乐部  我要评论(0)
  • 摘要:锁的理解锁产生于多线程并发应用,其作用是解决共享对象的同步同时也可以控制线程的行为。我认为锁不仅仅限于synchronize,ReentrantLock,ReadWriteLock.同时也包括CountDownLack,FutureTask,Semaphore,CyclicBarrier,Exchanger这些平时接触不多的并发控制类。后者经常会用在控制线程的运行行为。1.CountDownLack这种锁经常用来控制多个线程同时启动,并且能够及时感知这些线程是否全部运行结束。举例如下
  • 标签:Java 应用 Java并发

锁的理解

锁产生于多线程并发应用,其作用是解决共享对象的同步同时也可以控制线程的行为。我认为锁不仅仅限于synchronize,ReentrantLock,ReadWriteLock.同时也包括CountDownLack, FutureTask, Semaphore, CyclicBarrier, Exchanger这些平时接触不多的并发控制类。后者经常会用在控制线程的运行行为。

?

1.

CountDownLack 这种锁经常用来控制多个线程同时启动,并且能够及时感知这些线程是否全部运行结束。举例如下:

例子中m_begin用来为那10个线程发送启动指令,当m_begin.countDown()时,10个线程同时启动。同时m_end.await()这些线程结束的好消息。

class="java" name="code">/**
 * @filename CountDownLaunchTest.java
 * @date     2014-11-14 
 */
package lock;

import java.util.concurrent.CountDownLatch;

public class CountDownLatchTest
{
	public static void main(String[] args)
	{
		new CountDownLatchTest(10).runAll();
	}
	
	private int m_threadNum = 10;
	private CountDownLatch m_begin = null;
	private CountDownLatch m_end = null;
	
	CountDownLatchTest(int threadNum)
	{
		if(threadNum>0)
			m_threadNum = threadNum;
		m_begin = new CountDownLatch(1);
		m_end = new CountDownLatch(m_threadNum);
	}
	
	public void runAll()
	{
		for(int i=0; i<m_threadNum; i++)
		{
			new InnerThread(String.valueOf(i)).start();
		}
		
		m_begin.countDown();
		try
		{
			m_end.await();
		} catch (InterruptedException e)
		{
			e.printStackTrace();
		}
		
		System.out.println("all thread finished");
	}
	
	private class InnerThread extends Thread
	{
		private final String name;
		InnerThread(String threadName)
		{
			name = threadName;
			System.out.println(name+" inited");
		}
		@Override
		public void run()
		{
			try
			{
				//!!!!!注意是await,不是wait
				m_begin.await();
				Thread.currentThread().sleep(3000);
			} catch (InterruptedException e)
			{
				e.printStackTrace();
			}
			m_end.countDown();
			System.out.println(name+" finished");
		}
	}
}

??

2.FutureTask 这种锁可以获取一个线程任务的运行结果,也就是说我们有个任务需要启动一个线程进行处理,同时也需要得到这个线程的返回结果时,用这个锁比较好用。

/**
 * @filename FutureTaskTest.java
 * @date     2014-11-14 
 */
package lock;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class FutureTaskTest
{
	public static void main(String[] args)
	{
		new FutureTaskTest().test();
	}

	public void test()
	{
		FutureTask<Integer> futureTask = new FutureTask<Integer>(new InnerRunnable());
		futureTask.run();
		
		try
		{
			int result = futureTask.get();
			System.out.println(result);
		} catch (InterruptedException e)
		{
			e.printStackTrace();
		} catch (ExecutionException e)
		{
			//所有异常(除InterruptedException)均会封装成ExecutionException异常而抛出
			e.printStackTrace();
		}
	}
	
	private class InnerRunnable implements Callable<Integer>
	{
		public Integer call() throws Exception
		{
			try
			{
				Thread.sleep(3000);
//				throw new IllegalStateException();
			} catch (InterruptedException e)
			{
				e.printStackTrace();
				return -1;
			}
			return 0;
		}
	}
}

?

3.

Semaphore,信号量,其经常用在线程池数量的控制或者队列大小的控制上,根据预先设定好的数值,然后有线程来acquire()和release(),当申请的次数大于预设值时将阻塞,直到其他线程释放资源。

/**
 * @filename SemaphoreTest.java
 * @date     2014-11-14 
 */
package lock;

import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.Semaphore;

public class SemaphoreTest
{
	
	public static void main(String[] args)
	{
		SemaphoreTest test = new SemaphoreTest();
		test.init(10);
		test.releaseNumToPool(1);
		test.getNumFromPool();
		test.releaseNumToPool(11);
	}
	
	//通常用在连接池里面,用来限制申请连接的数目。
	private Semaphore sem = new Semaphore(0);
	
	private final ConcurrentSkipListSet<Integer> m_numPool = new ConcurrentSkipListSet<Integer>();
	
	public void init(int size)
	{
		for(int i=1; i<=size; i++)
			releaseNumToPool(i);
	}
	
	public Integer getNumFromPool()
	{
		try
		{
			sem.acquire();
		} catch (InterruptedException e)
		{
			e.printStackTrace();
		};
		if(m_numPool.isEmpty())
			return 0;
		return m_numPool.first();
	}
	
	public void releaseNumToPool(Integer num)
	{
		if(!m_numPool.contains(num))
		{
			m_numPool.add(num);
			sem.release();
		}
	}
}

?

4.

CyclicBarrier, 与CountDownLack和FutureTask不通的是,这种锁可以循环的使用,其用来规范线程任务的运行后行为,也就是当设置此锁有,多个线程运行完后,均会等在此锁上,当最后一个线程运行到此锁时,大家才继续运行。这个锁目前我能想到应用场景的地方不多,或许可以用在某些需要拼装工作同时拼装的场景里面吧。

/**
 * @filename CyclicBarrierTest.java
 * @date     2014-11-14 
 */
package lock;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest
{
	public static void main(String[] args)
	{
		CyclicBarrierTest test = new CyclicBarrierTest(10);
		
		for(int i=0; i<5; i++)
			test.startOnce();
	}
	
	private CyclicBarrier m_barrier;
	private final Object lock = new Object();
	public CyclicBarrierTest(int barrierSize)
	{
		m_barrier = new CyclicBarrier(barrierSize, new Runnable()
		{
			public void run()
			{
				synchronized (lock)
				{
					lock.notify();
				}
				System.out.println("all thread run completed");
			}
		});
	}
	
	
	public void startOnce()
	{
		for(int i=1,size=m_barrier.getParties(); i<=size; i++)
		{
			InnerThread thread = new InnerThread(i*1000, String.valueOf(i));
			thread.start();
		}

		synchronized (lock)
		{
			try
			{
				lock.wait();
			} catch (InterruptedException e)
			{
				e.printStackTrace();
			}
		}
			
	}
	
	private class InnerThread extends Thread
	{
		private long m_waitTime = 0;
		private String m_name;
		
		public InnerThread(long waitTime, String threadName)
		{
			if(waitTime>0)
				m_waitTime = waitTime;
			m_name = threadName;
		}
		
		@Override
		public void run()
		{
			try
			{
				Thread.sleep(m_waitTime);
				System.out.println(m_name+" waited "+m_waitTime);
				m_barrier.await();
			} catch (InterruptedException e)
			{
				e.printStackTrace();
			} catch (BrokenBarrierException e)
			{
				e.printStackTrace();
			}
			System.out.println(m_name+" completed");
		}
	}
}

?5.

Exchanger,其与SynchronizeQueue有相似之处,SynchronizeQueue是单向的,Exchanger是双向的。也就是说当两个线程运行到Exchanger时,双方均会将自己的数据交换给对方。目前我也没有想到很好的应用。

/**
 * @filename ExchangerTest.java
 * @date     2014-11-14 
 */
package lock;

import java.util.concurrent.Exchanger;

public class ExchangerTest
{
	public static void main(String[] args)
	{
		new ExchangerTest().start();
	}
	
	private Exchanger<String> exchanger = new Exchanger<String>();
	
	public void start()
	{
		new cargoThread().start();
		new moneyThread().start();
	}
	
	private class cargoThread extends Thread
	{
		String message = "cargo";
		
		public void run()
		{
			
			System.out.println("cargo thread is producing cargo");
			try
			{
				Thread.sleep(5000);
			} catch (InterruptedException e1)
			{
				e1.printStackTrace();
			}
			
			System.out.println("cargo thread begin to wait exchange");
			try
			{
				message = exchanger.exchange(message);
			} catch (InterruptedException e)
			{
				e.printStackTrace();
			}
			System.out.println("cargo thread exchange message "+message);
		}
	}
	
	private class moneyThread extends Thread
	{
		String message = "money";
		
		public void run()
		{
			System.out.println("money thread begin to wait exchange");
			try
			{
				message = exchanger.exchange(message);
			} catch (InterruptedException e)
			{
				e.printStackTrace();
			}
			System.out.println("money thread exchange message "+message);
		}
	}
	
	
}

?

上一篇: 节点与链表 下一篇: 没有下一篇了!
发表评论
用户名: 匿名