Java并发控制机制 Java并发控制机制详解

 public void method(){
synchronized(this){
// do something …
}
} 

 public void method(Object o){
// before
synchronized(o){
// do something ...
}
// after
} 

 synchronized(obj){
  while(<?>){
    obj.wait();
    // 收到通知后，继续执行。
  }
} 

 public class BlockQueue{
 private List list = new ArrayList();

 public synchronized Object pop() throws InterruptedException{
 while (list.size()==0){
 this.wait();
 }
 if (list.size()>0){
 return list.remove(0);
 } else{
 return null;
 }
 }

 public synchronized Object put(Object obj){
 list.add(obj);
 this.notify();
 }

} 

 try {
 if (lock.tryLock(5, TimeUnit.SECONDS)) { //如果已经被lock，尝试等待5s，看是否可以获得锁，如果5s后仍然无法获得锁则返回false继续执行
 // lock.lockInterruptibly();可以响应中断事件
 try { 
 //操作
 } finally {
 lock.unlock();
 }
 }
} catch (InterruptedException e) {
 e.printStackTrace(); //当前线程被中断时(interrupt)，会抛InterruptedException 
} 

 private ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private Lock readLock = readWriteLock.readLock();
private Lock writeLock = readWriteLock.writeLock();
public Object handleRead() throws InterruptedException {
  try {
    readLock.lock();
    Thread.sleep(1000);
    return value;
  }finally{
    readLock.unlock();
  }
}
public Object handleRead() throws InterruptedException {
  try {
    writeLock.lock();
    Thread.sleep(1000);
    return value;
  }finally{
    writeLock.unlock();
  }
} 

 public class ArrayBlockingQueue extends AbstractQueue
implements BlockingQueue, java.io.Serializable {
/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;

public ArrayBlockingQueue(int capacity, boolean fair) {
  if (capacity <= 0)
    throw new IllegalArgumentException();
  this.items = new Object[capacity];
  lock = new ReentrantLock(fair); 
  notEmpty = lock.newCondition(); // 生成与Lock绑定的Condition
  notFull = lock.newCondition();
}

public void put(E e) throws InterruptedException {
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
  lock.lockInterruptibly();
  try {
    while (count == items.length)
      notFull.await();
    insert(e);
  } finally {
    lock.unlock();
  }
}

private void insert(E x) {
  items[putIndex] = x;
  putIndex = inc(putIndex);
  ++count;
  notEmpty.signal(); // 通知
}

public E take() throws InterruptedException {
  final ReentrantLock lock = this.lock;
  lock.lockInterruptibly();
  try {
    while (count == 0) // 如果队列为空
      notEmpty.await(); // 则消费者队列要等待一个非空的信号
    return extract();
  } finally {
    lock.unlock();
  }
}

private E extract() {
  final Object[] items = this.items;
  E x = this.<E>cast(items[takeIndex]);
  items[takeIndex] = null;
  takeIndex = inc(takeIndex);
  --count;
  notFull.signal(); // 通知put() 线程队列已有空闲空间
  return x;
}

// other code
} 

 public class Pool {
private static final int MAX_AVAILABLE = 100;
private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);
public Object getItem() throws InterruptedException {
  available.acquire();
  // 申请一个许可
  // 同时只能有100个线程进入取得可用项，
  // 超过100个则需要等待
  return getNextAvailableItem();
}

public void putItem(Object x) {
  // 将给定项放回池内，标记为未被使用
  if (markAsUnused(x)) {
    available.release();
    // 新增了一个可用项，释放一个许可，请求资源的线程被激活一个
  }
}

// 仅作示例参考，非真实数据
protected Object[] items = new Object[MAX_AVAILABLE]; // 用于对象池复用对象
protected boolean[] used = new boolean[MAX_AVAILABLE]; // 标记作用

protected synchronized Object getNextAvailableItem() {
  for (int i = 0; i < MAX_AVAILABLE; ++i) {
    if (!used[i]) {
      used[i] = true;
      return items[i];
    }
  }
  return null;
}

protected synchronized boolean markAsUnused(Object item) {
  for (int i = 0; i < MAX_AVAILABLE; ++i) {
    if (item == items[i]) {
      if (used[i]) {
        used[i] = false;
        return true;
      } else {
        return false;
      }
    }
  }
  return false;
}
} 

 public class TestNum {
 // 通过匿名内部类覆盖ThreadLocal的initialValue()方法，指定初始值
 private static ThreadLocal seqNum = new ThreadLocal() {
 public Integer initialValue() {
 return 0;
 }
 };
 // 获取下一个序列值
 public int getNextNum() {
 seqNum.set(seqNum.get() + 1);
 return seqNum.get();
}public static void main(String[] args) {
 TestNum sn = new TestNum();
 //3个线程共享sn，各自产生序列号
 TestClient t1 = new TestClient(sn);
 TestClient t2 = new TestClient(sn);
 TestClient t3 = new TestClient(sn);
 t1.start();
 t2.start();
 t3.start();
 }
private static class TestClient extends Thread {
 private TestNum sn;
public TestClient(TestNum sn) {
 this.sn = sn;
 }
public void run() {
 for (int i = 0; i < 3; i++) {
 // 每个线程打出3个序列值
 System.out.println("thread[" + Thread.currentThread().getName() + "] --> sn["
 + sn.getNextNum() + "]");
 }
 }
 }
 } 

 public synchronized void syncMehod(){
beforeMethod();
mutexMethod();
afterMethod();
} 

 public void syncMehod(){
beforeMethod();
synchronized(this){
mutexMethod();
}
afterMethod();
} 

public class LinkedBlockingQueue extends AbstractQueue
implements BlockingQueue, java.io.Serializable {
/* Lock held by take, poll, etc /
private final ReentrantLock takeLock = new ReentrantLock();
/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();

/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();

public E take() throws InterruptedException {
  E x;
  int c = -1;
  final AtomicInteger count = this.count;
  final ReentrantLock takeLock = this.takeLock;
  takeLock.lockInterruptibly(); // 不能有两个线程同时读取数据
  try {
    while (count.get() == 0) { // 如果当前没有可用数据，一直等待put()的通知
      notEmpty.await();
    }
    x = dequeue(); // 从头部移除一项
    c = count.getAndDecrement(); // size减1
    if (c > 1)
      notEmpty.signal(); // 通知其他take()操作
  } finally {
    takeLock.unlock(); // 释放锁
  }
  if (c == capacity)
    signalNotFull(); // 通知put()操作，已有空余空间
  return x;
}

public void put(E e) throws InterruptedException {
  if (e == null) throw new NullPointerException();
  // Note: convention in all put/take/etc is to preset local var
  // holding count negative to indicate failure unless set.
  int c = -1;
  Node<E> node = new Node(e);
  final ReentrantLock putLock = this.putLock;
  final AtomicInteger count = this.count;
  putLock.lockInterruptibly(); // 不能有两个线程同时put数据
  try {
    /*
     * Note that count is used in wait guard even though it is
     * not protected by lock. This works because count can
     * only decrease at this point (all other puts are shut
     * out by lock), and we (or some other waiting put) are
     * signalled if it ever changes from capacity. Similarly
     * for all other uses of count in other wait guards.
     */
    while (count.get() == capacity) { // 队列满了 则等待
      notFull.await();
    }
    enqueue(node); // 加入队列
    c = count.getAndIncrement();// size加1
    if (c + 1 < capacity)
      notFull.signal(); // 如果有足够空间，通知其他线程
  } finally {
    putLock.unlock();// 释放锁
  }
  if (c == 0)
    signalNotEmpty();// 插入成功后，通知take()操作读取数据
}

// other code   
} 

 public void syncMehod(){
synchronized(lock){
method1();
}
synchronized(lock){
method2();
}
} 
JVM整合后的形式：
 public void syncMehod(){
synchronized(lock){
method1();
method2();
}
} 

public class TestAtomic {
private static final int MAX_THREADS = 3;
private static final int TASK_COUNT = 3;
private static final int TARGET_COUNT = 100 * 10000;
private AtomicInteger acount = new AtomicInteger(0);
private int count = 0;
synchronized int inc() {
  return ++count;
}

synchronized int getCount() {
  return count;
}

public class SyncThread implements Runnable {
  String name;
  long startTime;
  TestAtomic out;

  public SyncThread(TestAtomic o, long startTime) {
    this.out = o;
    this.startTime = startTime;
  }

  @Override
  public void run() {
    int v = out.inc();
    while (v < TARGET_COUNT) {
      v = out.inc();
    }
    long endTime = System.currentTimeMillis();
    System.out.println("SyncThread spend:" + (endTime - startTime) + "ms" + ", v=" + v);
  }
}

public class AtomicThread implements Runnable {
  String name;
  long startTime;

  public AtomicThread(long startTime) {
    this.startTime = startTime;
  }

  @Override
  public void run() {
    int v = acount.incrementAndGet();
    while (v < TARGET_COUNT) {
      v = acount.incrementAndGet();
    }
    long endTime = System.currentTimeMillis();
    System.out.println("AtomicThread spend:" + (endTime - startTime) + "ms" + ", v=" + v);
  }
}

@Test
public void testSync() throws InterruptedException {
  ExecutorService exe = Executors.newFixedThreadPool(MAX_THREADS);
  long startTime = System.currentTimeMillis();
  SyncThread sync = new SyncThread(this, startTime);
  for (int i = 0; i < TASK_COUNT; i++) {
    exe.submit(sync);
  }
  Thread.sleep(10000);
}

@Test
public void testAtomic() throws InterruptedException {
  ExecutorService exe = Executors.newFixedThreadPool(MAX_THREADS);
  long startTime = System.currentTimeMillis();
  AtomicThread atomic = new AtomicThread(startTime);
  for (int i = 0; i < TASK_COUNT; i++) {
    exe.submit(atomic);
  }
  Thread.sleep(10000);
}
}