Java中常见的并发控制手段浅析

public synchronized doProcess() {

     // 同步代码块，只会有一个线程执行

}

public class Single {

   private static volatile Single instance;

   private Single() {}

   public static Single getInstance() {

       if (instance == null ) {

           synchronized (Single. class ) {

               if (instance == null ) instance = new Single();

           }

       }

       return instance;

   }

}

public final int getAndIncrement() {

     return unsafe.getAndAddInt( this , valueOffset, 1 );

}

// unsafe 实现

// cas + 自选，不断的尝试更新设置，直到成功为止

public final int getAndAddInt(Object var1, long var2, int var4) {

     int var5;

     do {

         var5 = this .getIntVolatile(var1, var2);

     } while (! this 测试数据pareAndSwapInt(var1, var2, var5, var5 + var4));

     return var5;

}

private void doSome(ReentrantReadWriteLock.WriteLock writeLock) {

     try {

         writeLock.lock();

         System.out.println( "持有锁成功 " + Thread.currentThread().getName());

         Thread.sleep( 1000 );

         System.out.println( "执行完毕! " + Thread.currentThread().getName());

         writeLock.unlock();

     } catch (Exception e) {

         e.printStackTrace();

     }

}

@Test

public void lock() throws InterruptedException {

     ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();

     new Thread(()->doSome(reentrantReadWriteLock.writeLock())).start();

     new Thread(()->doSome(reentrantReadWriteLock.writeLock())).start();

     new Thread(()->doSome(reentrantReadWriteLock.writeLock())).start();

     Thread.sleep( 20000 );

}

AtomicInteger cnt = new AtomicInteger();

private void consumer(LinkedBlockingQueue<Integer> queue) {

     try {

         // 同步阻塞拿去数据

         int val = queue.take();

         Thread.sleep( 2000 );

         System.out.println( "成功拿到: " + val + " Thread: " + Thread.currentThread());

     } catch (InterruptedException e) {

         e.printStackTrace();

     } finally {

         // 添加数据

         System.out.println( "结束 " + Thread.currentThread());

         queue.offer(cnt.getAndAdd( 1 ));

     }

}

@Test

public void blockQueue() throws InterruptedException {

     LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>( 2 );

     queue.add(cnt.getAndAdd( 1 ));

     queue.add(cnt.getAndAdd( 1 ));

     new Thread(() -> consumer(queue)).start();

     new Thread(() -> consumer(queue)).start();

     new Thread(() -> consumer(queue)).start();

     new Thread(() -> consumer(queue)).start();

     Thread.sleep( 10000 );

}

private void semConsumer(Semaphore semaphore) {

     try {

         //同步阻塞，尝试获取信号

         semaphore.acquire( 1 );

         System.out.println( "成功拿到信号，执行: " + Thread.currentThread());

         Thread.sleep( 2000 );

         System.out.println( "执行完毕，释放信号: " + Thread.currentThread());

         semaphore.release( 1 );

     } catch (Exception e) {

         e.printStackTrace();

     }

}

@Test

public void semaphore() throws InterruptedException {

     Semaphore semaphore = new Semaphore( 2 );

     new Thread(() -> semConsumer(semaphore)).start();

     new Thread(() -> semConsumer(semaphore)).start();

     new Thread(() -> semConsumer(semaphore)).start();

     new Thread(() -> semConsumer(semaphore)).start();

     new Thread(() -> semConsumer(semaphore)).start();

     Thread.sleep(20_000);

}

@Test

public void countDown() throws InterruptedException {

     CountDownLatch countDownLatch = new CountDownLatch( 2 );

     new Thread(() -> {

         try {

             System.out.println( "do something in " + Thread.currentThread());

             Thread.sleep( 2000 );

         } catch (InterruptedException e) {

             e.printStackTrace();

         } finally {

             countDownLatch.countDown();

         }

     }).start();

     new Thread(() -> {

         try {

             System.out.println( "do something in t2: " + Thread.currentThread());

             Thread.sleep( 1000 );

         } catch (InterruptedException e) {

             e.printStackTrace();

         } finally {

             countDownLatch.countDown();

         }

     }).start();

     countDownLatch.await();

     System.out.printf( "结束" );

}

private void cyclicBarrierLogic(CyclicBarrier barrier, long sleep) {

     // 等待达到条件才放行

     try {

         System.out.println( "准备执行: " + Thread.currentThread() + " at: " + LocalDateTime.now());

         Thread.sleep(sleep);

         int index = barrier.await();

         System.out.println( "开始执行: " + index + " thread: " + Thread.currentThread() + " at: " + LocalDateTime.now());

     } catch (Exception e) {

         e.printStackTrace();

     }

}

@Test

public void testCyclicBarrier() throws InterruptedException {

     // 到达两个工作线程才能继续往后面执行

     CyclicBarrier barrier = new CyclicBarrier( 2 );

     // 三秒之后，下面两个线程的才会输出 开始执行

     new Thread(() -> cyclicBarrierLogic(barrier, 1000 )).start();

     new Thread(() -> cyclicBarrierLogic(barrier, 3000 )).start();

     Thread.sleep( 4000 );

     // 重置，可以再次使用

     barrier.reset();

     new Thread(() -> cyclicBarrierLogic(barrier, 1 )).start();

     new Thread(() -> cyclicBarrierLogic(barrier, 1 )).start();

     Thread.sleep( 10000 );

}

private void guavaProcess(RateLimiter rateLimiter) {

     try {

         // 同步阻塞方式获取

         System.out.println( "准备执行: " + Thread.currentThread() + " > " + LocalDateTime.now());

         rateLimiter.acquire();

         System.out.println( "执行中: " + Thread.currentThread() + " > " + LocalDateTime.now());

     } catch (Exception e) {

         e.printStackTrace();

     }

}

@Test

public void testGuavaRate() throws InterruptedException {

     // 1s 中放行两个请求

     RateLimiter rateLimiter = RateLimiter.create( 2 .0d);

     new Thread(() -> guavaProcess(rateLimiter)).start();

     new Thread(() -> guavaProcess(rateLimiter)).start();

     new Thread(() -> guavaProcess(rateLimiter)).start();

     new Thread(() -> guavaProcess(rateLimiter)).start();

     new Thread(() -> guavaProcess(rateLimiter)).start();

     new Thread(() -> guavaProcess(rateLimiter)).start();

     new Thread(() -> guavaProcess(rateLimiter)).start();

     Thread.sleep(20_000);

}