java之JUC系列-外部Tools-Executors|Semaphor|Exchanger|CyclicBarrier|CountDownLatch

简介:

前面写了两篇JDBC源码的文章,自己都觉得有点枯燥,先插一段JUC系列的文章来换换胃口,前面有文章大概介绍过J U C包含的东西,JUC体系包含的内容也是非常的多,不是一两句可以说清楚的,我这首先列出将会列举的JUC相关的内容,然后介绍本文的版本:Tools部分

J.U.C体系的主要大板块包含内容,如下图所示:



注意这个里面每个部分都包含很多的类和处理器,而且是相互包含,相互引用的,相互实现的。

说到J UC其实就是说java的多线程等和锁,前面说过一些状态转换,中断等,我们今天来用它的tools来实现一些有些小意思的东西,讲到其他内容的时候,再来想想这写tools是怎么实现的。

tools是本文说要讲到的重点,而tools主要包含哪些东西呢:


Tools也包含了5个部分的知识:ExecutorsSemaphor、Exchanger、CyclicBarrier、CountDownLatch,其实也就是五个工具类,这5个工具类有神马用途呢,就是我们接下来要将的内容了。


Executors:

其实它主要用来创建线程池,代理了线程池的创建,使得你的创建入口参数变得简单,通过方法名便知道了你要创建的线程池是什么样一个线程池,功能大概是什么样的,其实线程池内部都是统一的方法来实现,通过构造方法重载,使得实现不同的功能,但是往往这种方式很多时候不知道具体入口参数的改变有什么意思,除非读了源码才知道,此时builder模式的方式来完成,builder什么样的东西它告诉你就可以。

常见的方法有(都是静态方法):

1、创建一个指定大小的线程池,如果超过大小,放入blocken队列中,默认是LinkedBlockingQueue,默认的ThreadFactory为:Executors.defaultThreadFactory(),是一个Executors的一个内部类。

Executors.newFixedThreadPool(int) 

内部实现是:

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }


2、 创建一个指定大小的线程池,如果超过大小,放入blocken队列中,默认是LinkedBlockingQueue,自己指定ThreadFactory,自己写的ThreadFactory,必须implements ThreadFactory,实现方法:newThread(Runnable)

Executors.newFixedThreadPool(int,ThreadFactory) 
内部实现是:

    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

3、创建线程池长度为1的,也就是只有一个长度的线程池,多余的必须等待,它和调用 Executors.newFixedThreadPool(1)得到的结果一样:

 Executors.newSingleThreadExecutor() 

内部实现是:

    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
是不是蛮简单的,就是在变参数,你自己也可以new的。

4、和方法3类似,可以自定义ThreadFactory,这里就不多说了!

5、创建可以进行缓存的线程池,默认缓存60s,数据会放在一个SynchronousQueue上,而不会进入blocken队列中,也就是只要有线程进来就直接进入调度,这个不推荐使用,因为容易出问题,除非用来模拟一些并发的测试:

Executors.newCachedThreadPool();

内部实现为:

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
6、和方法5类似,增加自定义ThreadFactory

7、添加一个Schedule的调度器的线程池,默认只有一个调度:

Executors.newSingleThreadScheduledExecutor();

内部实现为(这里可以看到不是用ThreadPoolExector了,schedule换了一个类,内部实现通过ScheduledThreadPoolExecutor类里面的内部类ScheduledFutureTask来实现的,这个内部类是private,默认是引用不到的哦):

    public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
    }
8、和7一样,增加自己定义的ThreadFactory

9、添加一个schedule的线程池调度器,和newFixedThreadPool有点类似:

Executors.newScheduledThreadPool();
内部代码为:

    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

其实内部Exectors里面还有一些其他的方法,我们就不多说明了,另外通过这里,大家先可以了解一个大概,知道Exectors其实是一个工具类,提供一系列的静态方法,来完成对对应线程池的形象化创建,所以不用觉得很神奇,神奇的是内部是如何实现的,本文我们不阐述文章中各种线程池的实现,只是大概上有个认识,等到我们专门将 Exector系列的时候,我们会详细描述这些细节。

OK,我们继续下一个话题:

Semaphor,这个鸟东西是敢毛吃的呢?

答:通过名字就看出来了,是信号量。

信号量可以干什么呢?

答:根据一些阀值做访问控制。

OK,我们这里模拟一个当多个线程并发一段代码的时候,如何控制其访问速度:

import java.util.Random;
import java.util.concurrent.Semaphore;


  public class SemaphoreTest {
      private final static Semaphore MAX_SEMA_PHORE = new Semaphore(10);
      public static void main(String []args) {
           for(int i = 0 ; i < 100 ; i++) {
                final int num = i;
                final Random radom = new Random();
                new Thread() {
                     public void run() {
                         boolean acquired = false;
                         try {
                              MAX_SEMA_PHORE.acquire();
                              acquired = true;
                              System.out.println("我是线程:" + num + " 我获得了使用权!" + DateTimeUtil.getDateTime());
                              long time = 1000 * Math.max(1, Math.abs(radom.nextInt() % 10));
                              Thread.sleep(time);
                              System.out.println("我是线程:" + num + " 我执行完了!" + DateTimeUtil.getDateTime());
                         }catch(Exception e) {
                              e.printStackTrace();
                         }finally {
                              if(acquired) {
                                 MAX_SEMA_PHORE.release();
                              }
                         }
                      }
                }.start();
           }
      }
  }


这里是简单模拟并发100个线程去访问一段程序,此时要控制最多同时运行的是10个,用到了这个信号量,运行程序用了一个线程睡眠一个随机的时间来代替,你可以看到后面有线程说自己释放了,就有线程获得了,没释放是获取不到的,内部实现方面,我们暂时不管,暂时知道这样用就OK。


接下来:

Exchanger十个神马鬼东西呢?

答:线程之间交互数据,且在并发时候使用,两两交换,交换中不会因为线程多而混乱,发送出去没接收到会一直等,由交互器完成交互过程。

啥时候用,没想到案例?

答:的确很少用,而且案例很少,不过的确有这种案例,Exchanger

import java.util.concurrent.Exchanger;

public class ExchangerTest {
	
	public static void main(String []args) {
		final Exchanger <Integer>exchanger = new Exchanger<Integer>();
		for(int i = 0 ; i < 10 ; i++) {
			final Integer num = i;
			new Thread() {
				public void run() {
					System.out.println("我是线程:Thread_" + this.getName() + "我的数据是:" + num);
					try {
						Integer exchangeNum = exchanger.exchange(num);
						Thread.sleep(1000);
						System.out.println("我是线程:Thread_" + this.getName() + "我原先的数据为:" + num + " , 交换后的数据为:" + exchangeNum);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}.start();
		}
	}
}


这里运行你可以看到,如果某个线程和另一个线程传送了数据,它接受到的数据必然是另一个线程传递给他的,中间步骤由Exchanger去控制,其实你可以说,我自己随机取选择,不过中间的算法逻辑就要复杂一些了。


接下来:

CyclicBarrier,关卡模式,搞啥玩意的呢?

答:当你在很多环节需要卡住,要多个线程同时在这里都达到后,再向下走,很有用途。

能否举个例子,有点抽象?

答:团队出去旅行,大家一起先达到酒店住宿,然后一起达到游乐的地方游玩,然后一起坐车回家,每次需要点名后确认相关人员均达到,然后LZ一声令下,触发,大伙就疯子般的出发了。

下面的例子也是以旅游的方式来呈现给大家:

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

public class BarrierTest {
	
	private static final int THREAD_COUNT = 10;
	
	private final static CyclicBarrier CYCLIC_BARRIER = new CyclicBarrier(THREAD_COUNT  ,
		new Runnable() {
			public void run() {
				System.out.println("======>我是导游,本次点名结束,准备走下一个环节!");
			}
		}
	);
	
	public static void main(String []args) 
			throws InterruptedException, BrokenBarrierException {
		for(int i = 0 ; i < 10 ; i++) {
			new Thread(String.valueOf(i)) {
				public void run() {
					try {
						System.out.println("我是线程:" + this.getName() + " 我们达到旅游地点!");
						CYCLIC_BARRIER.await();
						System.out.println("我是线程:" + this.getName() + " 我开始骑车!");
						CYCLIC_BARRIER.await();
						System.out.println("我是线程:" + this.getName() + " 我们开始爬山!");
						CYCLIC_BARRIER.await();
						System.out.println("我是线程:" + this.getName() + " 我们回宾馆休息!");
						CYCLIC_BARRIER.await();
						System.out.println("我是线程:" + this.getName() + " 我们开始乘车回家!");
						CYCLIC_BARRIER.await();
						System.out.println("我是线程:" + this.getName() + " 我们到家了!");
					} catch (InterruptedException e) {
						e.printStackTrace();
					} catch (BrokenBarrierException e) {
						e.printStackTrace();
					}
				}
			}.start();
		}
	}
}

测试结果中可以发现,大家一起走到某个步骤后,导游说:“我是导游,本次点名结束,准备走下一个环节!”,然后才会进入下一个步骤,OK,这个有点意思吧,其实赛马也是这个道理,只是赛马通常只有一个步骤,所以我们还有一个方式是:

CountDownLatch的方式来完成赛马操作,CountDownLatch是用计数器来做的,所以它不可以被复用,如果要多次使用,就要从新new一个出来才可以。我们下面的代码中,用两组赛马,每组5个参与者来,做一个简单测试:

import java.util.concurrent.CountDownLatch;

public class CountDownLatchTest {
	
	private final static int GROUP_SIZE = 5;
	
	public static void main(String []args) {
		processOneGroup("分组1");
		processOneGroup("分组2");
	}
	
	private static void processOneGroup(final String groupName) {
		final CountDownLatch start_count_down = new CountDownLatch(1);
		final CountDownLatch end_count_down = new CountDownLatch(GROUP_SIZE);
		System.out.println("==========================>\n分组:" + groupName + "比赛开始:");
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			new Thread(String.valueOf(i)) {
				public void run() {
					System.out.println("我是线程组:【" + groupName + "】,第:" + this.getName() + " 号线程,我已经准备就绪!");
					try {
						start_count_down.await();//等待开始指令发出即:start_count_down.countDown();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("我是线程组:【" + groupName + "】,第:" + this.getName() + " 号线程,我已执行完成!");
					end_count_down.countDown();
				}
			}.start();
		}
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("各就各位,预备!");
		start_count_down.countDown();//开始赛跑
		try {
			end_count_down.await();//等待多个赛跑者逐个结束
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("分组:" + groupName + "比赛结束!");
	}
}


有点意思哈,如果你自己用多线程实现是不是有点麻烦,不过你可以用 Thread的join方法来实现,也就是线程的发生join的时候,当前线程(一般是主线程)要等到对应线程执行完run方法后才会进入下一步,为了模拟下,我们也来玩玩:

public class ThreadJoinTest {
	
	private final static int GROUP_SIZE = 5;

	public static void main(String []args) throws InterruptedException {
		Thread []threadGroup1 = new Thread[5];
		Thread []threadGroup2 = new Thread[5];
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			final int num = i;
			threadGroup1[i] = new Thread() {
				public void run() {
					int j = 0;
					while(j++ < 10) {
						System.out.println("我是1号组线程:" + num + " 这个是我第:" + j + " 次运行!");
					}
				}
			};
			threadGroup2[i] = new Thread() {
				public void run() {
					int j = 0;
					while(j++ < 10) {
						System.out.println("我是2号组线程:" + num + " 这个是我第:" + j + " 次运行!");
					}
				}
			};
			threadGroup1[i].start();
		}
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			threadGroup1[i].join();
		}
		System.out.println("-==================>线程组1执行完了,该轮到俺了!");
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			threadGroup2[i].start();
		}
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			threadGroup2[i].join();
		}
		System.out.println("全部结束啦!哈哈,回家喝稀饭!");
	}
}


代码是不是繁杂了不少,呵呵,我们再看看上面的信号量,如果不用工具,自己写会咋写,我们模拟CAS锁,使用Atomic配合完成咋来做呢。也来玩玩,呵呵:

import java.util.concurrent.atomic.AtomicInteger;

public class ThreadWaitNotify {
	
	private final static int THREAD_COUNT = 100;
	
	private final static int QUERY_MAX_LENGTH = 2;
	
	private final static AtomicInteger NOW_CALL_COUNT = new AtomicInteger(0);

	public static void main(String []args) throws InterruptedException {
		Thread []threads = new Thread[THREAD_COUNT];
		for(int i = 0 ; i < THREAD_COUNT ; i++) {
			threads[i] = new Thread(String.valueOf(i)) {
				synchronized public void run() {
					int nowValue = NOW_CALL_COUNT.get();
					while(true) {
						if(nowValue < QUERY_MAX_LENGTH && NOW_CALL_COUNT.compareAndSet(nowValue, nowValue + 1)) {
							break;//获取到了
						}
						try {
							this.wait(1000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						nowValue = NOW_CALL_COUNT.get();//获取一个数据,用于对比
					}
					System.out.println(this.getName() + "======我开始做操作了!");
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println(this.getName() + "======操作结束了!");
					NOW_CALL_COUNT.getAndDecrement();
					this.notify();
				}
			};
		}
		for(int i = 0 ; i < THREAD_COUNT ; i++) {
			threads[i].start();
		}
	}
}

还是有点意思哈,这样写就是大部分人对while循环那部分会写晕掉,主要是要不断去判定和尝试,wait()默认是长期等待,但是我们不想让他长期等待,就等1s然后再尝试,其实例子还可以改成wait一个随机的时间范围,这样模拟的效果会更加好一些;另外实际的代码中,如果获取到锁后,notify方法应当放在finally中,才能保证他肯定会执行notify这个方法。


OK,本文就是用,玩,希望玩得有点爽,我们后面会逐步介绍它的实现机制以及一写线程里头很好用,但是大家又不是经常用的东西。







目录
相关文章
|
3月前
|
安全 Java API
JAVA并发编程JUC包之CAS原理
在JDK 1.5之后,Java API引入了`java.util.concurrent`包(简称JUC包),提供了多种并发工具类,如原子类`AtomicXX`、线程池`Executors`、信号量`Semaphore`、阻塞队列等。这些工具类简化了并发编程的复杂度。原子类`Atomic`尤其重要,它提供了线程安全的变量更新方法,支持整型、长整型、布尔型、数组及对象属性的原子修改。结合`volatile`关键字,可以实现多线程环境下共享变量的安全修改。
|
21天前
|
Java
Java之CountDownLatch原理浅析
本文介绍了Java并发工具类`CountDownLatch`的使用方法、原理及其与`Thread.join()`的区别。`CountDownLatch`通过构造函数接收一个整数参数作为计数器,调用`countDown`方法减少计数,`await`方法会阻塞当前线程,直到计数为零。文章还详细解析了其内部机制,包括初始化、`countDown`和`await`方法的工作原理,并给出了一个游戏加载场景的示例代码。
Java之CountDownLatch原理浅析
|
3月前
|
算法 Java
JAVA并发编程系列(8)CountDownLatch核心原理
面试中的编程题目“模拟拼团”,我们通过使用CountDownLatch来实现多线程条件下的拼团逻辑。此外,深入解析了CountDownLatch的核心原理及其内部实现机制,特别是`await()`方法的具体工作流程。通过详细分析源码与内部结构,帮助读者更好地理解并发编程的关键概念。
|
2月前
|
存储 消息中间件 安全
JUC组件实战:实现RRPC(Java与硬件通过MQTT的同步通信)
【10月更文挑战第9天】本文介绍了如何利用JUC组件实现Java服务与硬件通过MQTT的同步通信(RRPC)。通过模拟MQTT通信流程,使用`LinkedBlockingQueue`作为消息队列,详细讲解了消息发送、接收及响应的同步处理机制,包括任务超时处理和内存泄漏的预防措施。文中还提供了具体的类设计和方法实现,帮助理解同步通信的内部工作原理。
JUC组件实战:实现RRPC(Java与硬件通过MQTT的同步通信)
|
3月前
|
存储 缓存 安全
【Java面试题汇总】多线程、JUC、锁篇(2023版)
线程和进程的区别、CAS的ABA问题、AQS、哪些地方使用了CAS、怎么保证线程安全、线程同步方式、synchronized的用法及原理、Lock、volatile、线程的六个状态、ThreadLocal、线程通信方式、创建方式、两种创建线程池的方法、线程池设置合适的线程数、线程安全的集合?ConcurrentHashMap、JUC
【Java面试题汇总】多线程、JUC、锁篇(2023版)
|
3月前
|
Java
JAVA并发编程系列(9)CyclicBarrier循环屏障原理分析
本文介绍了拼多多面试中的模拟拼团问题,通过使用 `CyclicBarrier` 实现了多人拼团成功后提交订单并支付的功能。与之前的 `CountDownLatch` 方法不同,`CyclicBarrier` 能够确保所有线程到达屏障点后继续执行,并且屏障可重复使用。文章详细解析了 `CyclicBarrier` 的核心原理及使用方法,并通过代码示例展示了其工作流程。最后,文章还提供了 `CyclicBarrier` 的源码分析,帮助读者深入理解其实现机制。
|
3月前
|
监控 Java 调度
【Java学习】多线程&JUC万字超详解
本文详细介绍了多线程的概念和三种实现方式,还有一些常见的成员方法,CPU的调动方式,多线程的生命周期,还有线程安全问题,锁和死锁的概念,以及等待唤醒机制,阻塞队列,多线程的六种状态,线程池等
161 6
【Java学习】多线程&JUC万字超详解
|
4月前
|
Java
Java 中 CyclicBarrier 和 CountDownLatch 的区别
【8月更文挑战第22天】
117 4
|
13天前
|
Java 开发者
Java多线程编程中的常见误区与最佳实践####
本文深入剖析了Java多线程编程中开发者常遇到的几个典型误区,如对`start()`与`run()`方法的混淆使用、忽视线程安全问题、错误处理未同步的共享变量等,并针对这些问题提出了具体的解决方案和最佳实践。通过实例代码对比,直观展示了正确与错误的实现方式,旨在帮助读者构建更加健壮、高效的多线程应用程序。 ####
|
4天前
|
缓存 Java 开发者
Java多线程编程的陷阱与最佳实践####
本文深入探讨了Java多线程编程中常见的陷阱,如竞态条件、死锁和内存一致性错误,并提供了实用的避免策略。通过分析典型错误案例,本文旨在帮助开发者更好地理解和掌握多线程环境下的编程技巧,从而提升并发程序的稳定性和性能。 ####