java一些常用并发工具示例

/**

* 闭锁测试（菩提树下的杨过 http://yjmyzz.cnblogs.com/）

*

* @throws InterruptedException

*/

@Test

publicvoidcountdownLatch()throwsInterruptedException {

CountDownLatch startLatch =newCountDownLatch(1);//类似发令枪

CountDownLatch endLatch =newCountDownLatch(10);//这里的数量，要与线程数相同

for(inti =0; i <10; i++) {

Thread t =newThread(() -> {

try{

startLatch.await();//先等着，直到发令枪响，防止有线程先run

System.out.println(Thread.currentThread().getName() +" is running...");

Thread.sleep(10);

}catch(InterruptedException e) {

Thread.currentThread().interrupt();

}finally{

endLatch.countDown();//每个线程执行完成后，计数

}

});

t.setName("线程-"+ i);

t.start();

}

longstart = System.currentTimeMillis();

startLatch.countDown();//发令枪响，所有线程『开跑』

endLatch.await();//等所有线程都完成

longend = System.currentTimeMillis();

System.out.println("done! exec time => "+ (end - start) +" ms");

}

publicclassBoundedHashSet<T> {

privatefinalSet<T> set;

privatefinalSemaphore semaphore;

publicBoundedHashSet(intbound) {

this.set = Collections.synchronizedSet(newHashSet<T>());

this.semaphore =newSemaphore(bound);

}

publicbooleanadd(T t)throwsInterruptedException {

if(!semaphore.tryAcquire(5, TimeUnit.SECONDS)) {

returnfalse;

}

;

booleanadded =false;

try{

added = set.add(t);

returnadded;

}finally{

if(!added) {

semaphore.release();

}

}

}

publicbooleanremove(Object o) {

booleanremoved = set.remove(o);

if(removed) {

semaphore.release();

}

returnremoved;

}

}

@Test

publicvoidsemaphoreTest()throwsInterruptedException {

BoundedHashSet<String> set =newBoundedHashSet<>(5);

for(inti =0; i <6; i++) {

if(set.add(i +"")) {

System.out.println(i +" added !");

}else{

System.out.println(i +" not add to Set!");

}

}

}

publicclassWorkerextendsThread {

privateCyclicBarrier cyclicBarrier;

publicWorker(CyclicBarrier cyclicBarrier) {

this.cyclicBarrier = cyclicBarrier;

}

privatevoidstep1() {

System.out.println(this.getName() +" step 1 ...");

}

privatevoidstep2() {

System.out.println(this.getName() +" step 2 ...");

}

publicvoidrun() {

step1();

try{

cyclicBarrier.await();

}catch(InterruptedException e) {

e.printStackTrace();

}catch(BrokenBarrierException e) {

e.printStackTrace();

}

step2();

}

}

@Test

publicvoidcyclicBarrierTest()throwsInterruptedException, BrokenBarrierException {

CyclicBarrier cyclicBarrier =newCyclicBarrier(11);

for(inti =0; i <10; i++) {

Worker w =newWorker(cyclicBarrier);

w.start();

}

cyclicBarrier.await();

}

@Test

publicvoidexchangerTest() {

Exchanger<String> exchanger =newExchanger<>();

Thread t1 =newThread(() -> {

String temp ="AAAAAA";

System.out.println("thread 1 交换前："+ temp);

try{

temp = exchanger.exchange(temp);

}catch(InterruptedException e) {

e.printStackTrace();

}

System.out.println("thread 1 交换后："+ temp);

});

Thread t2 =newThread(() -> {

String temp ="BBBBBB";

System.out.println("thread 2 交换前："+ temp);

try{

temp = exchanger.exchange(temp);

}catch(InterruptedException e) {

e.printStackTrace();

}

System.out.println("thread 2 交换后："+ temp);

});

t1.start();

t2.start();

}

@Test

publicvoidfutureTaskTest()throwsExecutionException, InterruptedException, TimeoutException {

Callable<String> callable = () -> {

System.out.println("很耗时的操作处理中。。。");

Thread.sleep(5000);

return"done";

};

FutureTask<String> futureTask =newFutureTask<>(callable);

System.out.println("就绪。。。");

newThread(futureTask).start();

System.out.println("主线程其它处理。。。");

System.out.println(futureTask.get());

System.out.println("处理完成！");

System.out.println("-----------------");

System.out.println("executor 就绪。。。");

ExecutorService executorService = Executors.newSingleThreadExecutor();

Future<String> future = executorService.submit(callable);

System.out.println(future.get(10, TimeUnit.SECONDS));

}

@Test

publicvoidblockingQueueTest()throwsInterruptedException {

finalBlockingQueue<String> blockingDeque =newArrayBlockingQueue<>(5);

Thread producer =newThread() {

publicvoidrun() {

Random rnd =newRandom();

while(true) {

try{

inti = rnd.nextInt(10000);

blockingDeque.put(i +"");

System.out.println(this.getName() +" 产生了一个数字："+ i);

Thread.sleep(rnd.nextInt(50));//模拟生产者快速生产

}catch(InterruptedException e) {

Thread.currentThread().interrupt();

}

}

}

};

producer.setName("producer 1");

Thread consumer =newThread() {

publicvoidrun() {

while(true) {

Random rnd =newRandom();

try{

String i = blockingDeque.take();

System.out.println(this.getName() +" 消费了一个数字："+ i);

Thread.sleep(rnd.nextInt(10000));//消费者模拟慢速消费

}catch(InterruptedException e) {

Thread.currentThread().interrupt();

}

}

}

};

consumer.setName("consumer 1");

producer.start();

consumer.start();

while(true) {

Thread.sleep(100);

}

}