长文慎入-探索Java并发编程与高并发解决方案

/**

• 请求总数
  */
  public static int clientTotal = 5000;

/**

• 同时并发执行的线程数
  */
  public static int threadTotal = 200;

/**

• 工作内存
  */
  public static AtomicLong count = new AtomicLong(0);

public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(threadTotal);
final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
for (int i = 0; i < clientTotal ; i++) {
executorService.execute(() -> {
try {
System.out.println();
semaphore.acquire();
add();
semaphore.release();
} catch (Exception e) {
log.error("exception", e);
}
countDownLatch.countDown();
});
}
countDownLatch.await();
executorService.shutdown();
//主内存
log.info("count:{}", count.get());
}

private static void add() {
count.incrementAndGet();
// count.getAndIncrement();
}
}

/**

• 请求总数
  */
  public static int clientTotal = 5000;

/**

• 同时并发执行的线程数
  */
  public static int threadTotal = 200;

public static int count = 0;

public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
final Semaphore semaphore = new Semaphore(threadTotal);
final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
for (int i = 0; i < clientTotal ; i++) {
executorService.execute(() -> {
try {
semaphore.acquire();
add();
semaphore.release();
} catch (Exception e) {
log.error("exception", e);
}
countDownLatch.countDown();
});
}
countDownLatch.await();
executorService.shutdown();
log.info("count:{}", count);
}

private synchronized static void add() {
count++;
}
}

synchronized 修正计数类方法
- 修饰类:括号括起来的部分，作用于所有对象
子类继承父类的被 synchronized 修饰方法时,是没有 synchronized 修饰的!!!

/**

• 私有构造函数
  */
  private SingletonExample4() {

}

// 1、memory = allocate() 分配对象的内存空间
// 2、ctorInstance() 初始化对象
// 3、instance = memory 设置instance指向刚分配的内存

// JVM和cpu优化，发生了指令重排

// 1、memory = allocate() 分配对象的内存空间
// 3、instance = memory 设置instance指向刚分配的内存
// 2、ctorInstance() 初始化对象

/**

• 单例对象
  */
  private static SingletonExample4 instance = null;

/**

• 静态的工厂方法
• @return
  */
  public static SingletonExample4 getInstance() {
  // 双重检测机制 // B
  if (instance == null) {
  // 同步锁
  synchronized (SingletonExample4.class) {
  if (instance == null) {
  // A - 3
  instance = new SingletonExample4();
  }
  }
  }
  return instance;
  }
  }

  ![](https://upload-images.jianshu.io/upload_images/4685968-823166cdf7936293.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
  ![](https://upload-images.jianshu.io/upload_images/4685968-9002671b71096f6c.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
  #7 AQS
  ##7.1 介绍
  ![数据结构](https://upload-images.jianshu.io/upload_images/4685968-918dcaea77d556e9.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
  - 使用Node实现FIFO队列，可以用于构建锁或者其他同步装置的基础框架
  - 利用了一个int类型表示状态
  - 使用方法是继承
  - 子类通过继承并通过实现它的方法管理其状态{acquire 和release} 的方法操纵状态
  - 可以同时实现排它锁和共享锁模式(独占、共享)
  同步组件
  ###CountDownLatch

  package com.mmall.concurrency.example.aqs;

private final static int threadCount = 200;

public static void main(String[] args) throws Exception {

ExecutorService exec = Executors.newCachedThreadPool();

final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

for (int i = 0; i < threadCount; i++) {
    final int threadNum = i;
    exec.execute(() -> {
        try {
            test(threadNum);
        } catch (Exception e) {
            log.error("exception", e);
        } finally {
            countDownLatch.countDown();
        }
    });
}
countDownLatch.await();
log.info("finish");
exec.shutdown();

}

private static void test(int threadNum) throws Exception {
Thread.sleep(100);
log.info("{}", threadNum);
Thread.sleep(100);
}
}

package com.mmall.concurrency.example.aqs;

private final static int threadCount = 200;

public static void main(String[] args) throws Exception {

ExecutorService exec = Executors.newCachedThreadPool();

final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

for (int i = 0; i < threadCount; i++) {
    final int threadNum = i;
    exec.execute(() -> {
        try {
            test(threadNum);
        } catch (Exception e) {
            log.error("exception", e);
        } finally {
            countDownLatch.countDown();
        }
    });
}
countDownLatch.await(10, TimeUnit.MILLISECONDS);
log.info("finish");
exec.shutdown();

}

private static void test(int threadNum) throws Exception {
Thread.sleep(100);
log.info("{}", threadNum);
}
}

##Semaphore用法
![](https://upload-images.jianshu.io/upload_images/4685968-e6cbcd4254c642c5.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
![](https://upload-images.jianshu.io/upload_images/4685968-dbefbf2c76ad5a2a.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
![](https://upload-images.jianshu.io/upload_images/4685968-41f5f5a5fd135804.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
##CycliBarrier

package com.mmall.concurrency.example.aqs;

private static CyclicBarrier barrier = new CyclicBarrier(5);

public static void main(String[] args) throws Exception {

ExecutorService executor = Executors.newCachedThreadPool();

for (int i = 0; i < 10; i++) {
    final int threadNum = i;
    Thread.sleep(1000);
    executor.execute(() -> {
        try {
            race(threadNum);
        } catch (Exception e) {
            log.error("exception", e);
        }
    });
}
executor.shutdown();

}

private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
barrier.await();
log.info("{} continue", threadNum);
}
}

![](https://upload-images.jianshu.io/upload_images/4685968-4fb51fa4926fd70e.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

package com.mmall.concurrency.example.aqs;

private static CyclicBarrier barrier = new CyclicBarrier(5);

public static void main(String[] args) throws Exception {

ExecutorService executor = Executors.newCachedThreadPool();

for (int i = 0; i < 10; i++) {
    final int threadNum = i;
    Thread.sleep(1000);
    executor.execute(() -> {
        try {
            race(threadNum);
        } catch (Exception e) {
            log.error("exception", e);
        }
    });
}
executor.shutdown();

}

private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
try {
barrier.await(2000, TimeUnit.MILLISECONDS);
} catch (Exception e) {
log.warn("BarrierException", e);
}
log.info("{} continue", threadNum);
}
}

![await 超时导致程序抛异常](https://upload-images.jianshu.io/upload_images/4685968-0f899c23531f8ee8.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)

package com.mmall.concurrency.example.aqs;

private final static int threadCount = 20;

public static void main(String[] args) throws Exception {

ExecutorService exec = Executors.newCachedThreadPool();

final Semaphore semaphore = new Semaphore(3);

for (int i = 0; i < threadCount; i++) {
    final int threadNum = i;
    exec.execute(() -> {
        try {
            // 尝试获取一个许可
            if (semaphore.tryAcquire()) {
                test(threadNum);
                // 释放一个许可
                semaphore.release();
            }
        } catch (Exception e) {
            log.error("exception", e);
        }
    });
}
exec.shutdown();

}

private static void test(int threadNum) throws Exception {
log.info("{}", threadNum);
Thread.sleep(1000);
}