首页 文章 精选 留言 我的

精选列表

搜索[源码],共10005篇文章
优秀的个人博客,低调大师

ConcurrentHashMap 源码分析

和 HashMap 不同的是,ConcurrentHashMap 采用分段加锁的方式保障线程安全,JDK 1.8 之后,ConcurrentHashMap 的底层数据结构从 1.8 开始跟 HashMap 差不多。 HashTable 也是线程安全的,存储 Key-Value 键值对的数据结构,Key 和 Value 都不能为空,但不推荐使用,因为其所有的方法采用 synchronized 修饰,效率低。 Key 和 Value 都不能为 Null 的原因是:如果 map.get(key) 返回 null,可以认为是 value 的值本来就是 null,也可以认为 map 中不存在 key 的存储数据,因此具有二义性,但 HashMap 在单线程环境,可以通过 map.containsKey(key) 判断,消除而已性。 但在多线程环境中,map.get(key) 和 map.containsKey(key) 是非原子的操作,可能在线程 A 的两个语句运行之间,其他线程 B 运行 map.put(key,value),导致线程 A 无法消除上面的二义性。 参考 https://www.cnblogs.com/thisiswhy/p/12059240.html 下图是 ConcurrentHashMap 的 UML 关系图。 1、底层存储结构 1.1、JDK 1.7 的存储结构,了解即可 在 JDK 1.7 ,ConcurrentHashMap 通过对 Segment 的分段加锁实现线程安全。一个 Segment 里面就是 HashMap 的存储结构,可以扩容。Segment 的数据量初始化以后不可以更改,默认值 16,因此默认支持 16 个线程同时操作 ConcurrentHashMap。 1.2 JDK 1.8 的存储结构 JDK 1.8 之后,存储结构变化比较大,跟 HashMap 类似。红黑树节点小于某个数(默认值 6) 又会转换为链表。 [ ] ConcurrentHashMap的主要成员变量,类似 HashMap,补上注释 2、ConcurrentHashMap 的构造方法 ConcurrentHashMap 的默认构造容量为 16,在初始化的时候并不会初始化 table 数组。同 HashMap 一样,在 put 第一个元素的时候才会 initTable() 初始化数组。 /** Creates a new, empty map with the default initial table size (16). */ public ConcurrentHashMap() { } // 设置初始化大小的构造函数 public ConcurrentHashMap(int initialCapacity) { this(initialCapacity, LOAD_FACTOR, 1); } // 根据传入的 map 初始化 public ConcurrentHashMap(Map<? extends K, ? extends V> m) { this.sizeCtl = DEFAULT_CAPACITY; putAll(m); } // 设置初始容量和加载因子的大小 public ConcurrentHashMap(int initialCapacity, float loadFactor) { this(initialCapacity, loadFactor, 1); } // 初始容量、加载因子、并发级别 public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { // 数据校验 if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); // 如果初始容量小于并发级别 if (initialCapacity < concurrencyLevel) // Use at least as many bins initialCapacity = concurrencyLevel; // as estimated threads // 一些比较 long size = (long)(1.0 + (long)initialCapacity / loadFactor); int cap = (size >= (long)MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int)size); this.sizeCtl = cap; } 3、get、put 方法 3.1 get 方法,根据 key 找 value,没有返回 null get 的流程总体和 HashMap 差不多,只不过是通过头结点的 hash 值判断是红黑树还是链表。 static final int MOVED = -1; // 转发节点? TODO 作用? static final int TREEBIN = -2; // 跟节点 static final int RESERVED = -3; // 临时保留的节点? TODO 作用? static final int HASH_BITS = 0x7fffffff; // hash 的扰动函数 spread() 计算用的 // 根据 key 获取 value 值 public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; // 计算 hash 值 int h = spread(key.hashCode()); // 集散所在的 hash 桶 if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { if ((eh = e.hash) == h) { // 头结点,刚好是要找的节点 if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } else if (eh < 0) // 头结点 hash 值小于 0,说明正在扩容或者是红黑树,find 查找 return (p = e.find(h, key)) != null ? p.val : null; while ((e = e.next) != null) { // 链表遍历查找 if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; } 3.2、put 方法 put 方法的流程跟 HashMap 的流程差不多,不同点在于线程安全,自旋,CAS,synchronized onlyIfAbsent 如果为 true ,如果已经存在了 key,不会替换旧的值。 public V put(K key, V value) { return putVal(key, value, false); } /** Implementation for put and putIfAbsent */ final V putVal(K key, V value, boolean onlyIfAbsent) { // key 和 value 都不能为 null if (key == null || value == null) throw new NullPointerException(); // 计算 hash(key) 的扰动函数 int hash = spread(key.hashCode()); // 离岸边的长度 int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; K fk; V fv; // 如果 table 还没有初始化,就初始化 table (自旋+CAS) if (tab == null || (n = tab.length) == 0) tab = initTable(); else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 如果当前 hash 桶为 null,直接放入,CAS 加入,成功了就直接 break if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value))) break; // no lock when adding to empty bin } // TODO : else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else if (onlyIfAbsent // check first node without acquiring lock && fh == hash && ((fk = f.key) == key || (fk != null && key.equals(fk))) && (fv = f.val) != null) return fv; else { // 旧的值 V oldVal = null; // 加锁 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; // 如果存在 hash(key) 和 key 对应的节点,直接更改 value 值 if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } Node<K,V> pred = e; if ((e = e.next) == null) { // 不存在直接放入,因为前面加锁了 pred.next = new Node<K,V>(hash, key, value); break; } } } // 如果是红黑树,红黑树插入 else if (f instanceof TreeBin) { Node<K,V> p; binCount = 2; if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } else if (f instanceof ReservationNode) throw new IllegalStateException("Recursive update"); } } if (binCount != 0) { // 是否要转为红黑树 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); // 旧的值 if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); return null; } 4、TODO ConcurrentHashMap 的扩容方法 ConcurrentHashMap 也是默认扩容 2 倍,扩容的方法 transfer() Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; 5、总结 ConcurrentHashMap 在 JDK 1.7 和 1.8 变化很大,在 JDK 1.7 中,采用 Segment 分段存储数据,也通过 Segment 分段加锁。 而在 JDK 1.8 中,使用 synchronized 锁定 hash 桶的链表的首节点/红黑树的根节点,只要 hash(key) 不冲突,就不会影响其他线程。

优秀的个人博客,低调大师

ArrayList源码分析

一、核心变量 // 序列化ID private static final long serialVersionUID = 8683452581122892189L; // 默认初始化容量 private static final int DEFAULT_CAPACITY = 10; // 空数组 private static final Object[] EMPTY_ELEMENTDATA = {}; // 空数组 private static final Object[] DEFAULTCAPACITY_EMPTY_ELEMENTDATA = {}; // 存储数据元素的数组 transient Object[] elementData; // non-private to simplify nested class access // 当前arraylist集合的大小,也就是elementData数组中数据元素的个数 private int size; 二、构造函数 /** * 一:无参构造方法 */ public ArrayList() { this.elementData = DEFAULTCAPACITY_EMPTY_ELEMENTDATA; } /** * 二:携带一个int类型的参数,指定arraylist的初始容量 */ public ArrayList(int initialCapacity) { if (initialCapacity > 0) { this.elementData = new Object[initialCapacity]; } else if (initialCapacity == 0) { this.elementData = EMPTY_ELEMENTDATA; } else { throw new IllegalArgumentException("Illegal Capacity: "+ initialCapacity); } } 1、无参构造 可以看到,在构造方法中直接将 elementData 指向 DEFAULTCAPACITY_EMPTY_ELEMENTDATA空数组,这个时候该ArrayList的size为初始值0。 1、有参构造 进行参数校验: 参数大于0,则指定数组长度; 参数等于0,则为空数组; 参数小于0,则抛异常。 三、add方法 /** * 一:直接添加数据元素到arraylist的尾部 */ public boolean add(E e) { //是否扩容、记录modCount ensureCapacityInternal(size + 1); // Increments modCount!! //把值添加到数组尾部 elementData[size++] = e; return true; } ---------------------------------------------------------------------- private void ensureCapacityInternal(int minCapacity) { //minCapacity=size+1; //minCapacity表示如果添加成功后,数组的最小长度 //如果为无参构造 if (elementData == DEFAULTCAPACITY_EMPTY_ELEMENTDATA) { //取默认长度和minCapacity的最大值,即10 minCapacity = Math.max(DEFAULT_CAPACITY, minCapacity); } //是否扩容 ensureExplicitCapacity(minCapacity); } ---------------------------------------------------------------------- private void ensureExplicitCapacity(int minCapacity) { modCount++; // 如果添加后最小长度大于 数组长度 if (minCapacity - elementData.length > 0) //扩容 grow(minCapacity); } ---------------------------------------------------------------------- private void grow(int minCapacity) { //获取数组长度 int oldCapacity = elementData.length; //1.5倍扩容 int newCapacity = oldCapacity + (oldCapacity >> 1); //1.5倍扩容也不够用 if (newCapacity - minCapacity < 0) //扩容后长度=minCapacity newCapacity = minCapacity; //简直最大长度 if (newCapacity - MAX_ARRAY_SIZE > 0) newCapacity = hugeCapacity(minCapacity); // 复制 elementData = Arrays.copyOf(elementData, newCapacity); } 代码中已经有注释了,很清晰。 四、remove方法 /* * 一:根据角标进行remove操作 */ public E remove(int index) { // 1. 对角标越界进行判断 if (index >= size) throw new IndexOutOfBoundsException(outOfBoundsMsg(index)); // 2.modCount自增1 modCount++; // 3.获取到指定下角标位置的数据 E oldValue = (E) elementData[index]; // 4.计算需要移动的元素个数 int numMoved = size - index - 1; if (numMoved > 0) // 5. 指定角标位置后的元素前移一位,效率低 System.arraycopy(elementData, index+1, elementData, index, numMoved); // 6.将size自减1,并将数组末尾置为null,便于垃圾回收 elementData[--size] = null; // clear to let GC do its work // 7.最后将所要删除的数据元素return掉 return oldValue; } /* * 二:根据数据元素进行remove操作 */ public boolean remove(Object o) { if (o == null) { for (int index = 0; index < size; index++) if (elementData[index] == null) { fastRemove(index); return true; } } else { for (int index = 0; index < size; index++) if (o.equals(elementData[index])) { fastRemove(index); return true; } } return false; } ---------------------------------------------------------------------- private void fastRemove(int index) { // 1.modCount的值自增1 modCount++; // 2.计算需要移动的元素个数 int numMoved = size - index - 1; if (numMoved > 0) // 3. 指定角标位置后的元素前移一位 System.arraycopy(elementData, index+1, elementData, index, numMoved); // 4.将size自减1,并将数组末尾置为null,便于垃圾回收 elementData[--size] = null; // clear to let GC do its work } 五、set方法 public void set(E e) { if (lastRet < 0) throw new IllegalStateException(); //检查modCount checkForComodification(); try { ArrayList.this.set(lastRet, e); } catch (IndexOutOfBoundsException ex) { throw new ConcurrentModificationException(); } } ---------------------------------------------------------------------- public E set(int index, E element) { rangeCheck(index); E oldValue = elementData(index); elementData[index] = element; return oldValue; } ---------------------------------------------------------------------- final void checkForComodification() { if (expectedModCount != ArrayList.this.modCount) throw new ConcurrentModificationException(); } 六、get方法 public E get(int index) { rangeCheck(index); checkForComodification(); return ArrayList.this.elementData(offset + index); } ---------------------------------------------------------------------- E elementData(int index) { return (E) elementData[index]; } 七、clear方法 public void clear() { modCount++; // clear to let GC do its work for (int i = 0; i < size; i++) elementData[i] = null; size = 0; } 八、contains方法 public boolean contains(Object o) { return indexOf(o) >= 0; } ---------------------------------------------------------------------- public int indexOf(Object o) { if (o == null) { for (int i = 0; i < size; i++) if (elementData[i]==null) return i; } else { for (int i = 0; i < size; i++) if (o.equals(elementData[i])) return i; } return -1; } 未命名文件.jpg 九、fail-fast机制 ail-fast机制是集合中的一种错误检测机制,我们在操作集合中经常会遇到 java.util.ConcurrentModificationException异常,产生该异常的原因就是fail-fast机制。 实现:如果在迭代期间计数器被修改,那么hasNext或next将抛出concurrentModificationException 缺点:这种检查是没有同步的情况下进行的,因此可能会看到失效的计数值,而迭代器可能并没有意识到已经发生了修改。这是一种设计上的权衡,从而降低了并发修改操作的检测代码对程序性能带来的影响。 十、可能的并发问题 add() EX:a、100个元素,可能最后数组长度不到100。 两个线程并发add,对索引位置5的地方几乎同时赋值,第二个线程会覆盖第一个线程的值,并且size少了1个。 b、假设有两个线程在操作同一个ArrayList,线程一执行step1(容量足够)后被挂起,线程二执行add()方法后,线程一被唤醒,这时线程一因为已经不再判断容量是否足够(已经判断过),执行step2就会出现数组越界 数组容量检测的并发问题 remove 两个线程有可能会想要删除同一个内容,一个线程先完成的时候第二个线程再删,就找不到这个内容了

优秀的个人博客,低调大师

Zuul源码阅读

Motivation 说一下,为什么要阅读Netflix Zuul,最近在看alibaba/Sentinel 在网关方面的应用。发现网关的设计模式有很多通用的地方。Netflix Zuul 对比Zuul2 更能直接的看出网关的主要功能和核心设计,易于上手,所以选择这个。同时也再次实践一下怎么高效阅读源代码。 Design and Function Zuul 的主要功能,包括,鉴权,路由,流量监控,负载,实时响应(动态配置和开关),错误处理。 这张图就是根据这些功能提出的设计,通过Filter 将整个过程连接起来, 可扩展性: 在不同阶段做不同的处理,做隔离。 通过groovy 脚本加载来实现动态加载新的filter。 ZuulFilter 下面我们就来看总重要的 ZuulFilter 的设计吧,首先是类结构图 先看最基本的IZuulFilter 接口, 定义了两个方法,这个filter 要不要执行,怎么执行。 public interface IZuulFilter { /** * a "true" return from this method means that the run() method should be invoked * * @return true if the run() method should be invoked. false will not invoke the run() method */ boolean shouldFilter(); /** * if shouldFilter() is true, this method will be invoked. this method is the core method of a ZuulFilter * * @return Some arbitrary artifact may be returned. Current implementation ignores it. * @throws ZuulException if an error occurs during execution. */ Object run() throws ZuulException; } 这里暂时提出疑问,throw ZuulException 会怎样?,返回任意result 怎么处理。 下面来看 Abstratct ZuulFilter类: /** * Base abstract class for ZuulFilters. The base class defines abstract methods to define: * filterType() - to classify a filter by type. Standard types in Zuul are "pre" for pre-routing filtering, * "route" for routing to an origin, "post" for post-routing filters, "error" for error handling. * We also support a "static" type for static responses see StaticResponseFilter. * Any filterType made be created or added and run by calling FilterProcessor.runFilters(type) * <p/> * filterOrder() must also be defined for a filter. Filters may have the same filterOrder if precedence is not * important for a filter. filterOrders do not need to be sequential. * <p/> * ZuulFilters may be disabled using Archius Properties. * <p/> * By default ZuulFilters are static; they don't carry state. This may be overridden by overriding the isStaticFilter() property to false * * @author Mikey Cohen * Date: 10/26/11 * Time: 4:29 PM */ public abstract class ZuulFilter implements IZuulFilter, Comparable<ZuulFilter> { private final AtomicReference<DynamicBooleanProperty> filterDisabledRef = new AtomicReference<>(); /** * to classify a filter by type. Standard types in Zuul are "pre" for pre-routing filtering, * "route" for routing to an origin, "post" for post-routing filters, "error" for error handling. * We also support a "static" type for static responses see StaticResponseFilter. * Any filterType made be created or added and run by calling FilterProcessor.runFilters(type) * * @return A String representing that type */ abstract public String filterType(); /** * filterOrder() must also be defined for a filter. Filters may have the same filterOrder if precedence is not * important for a filter. filterOrders do not need to be sequential. * * @return the int order of a filter */ abstract public int filterOrder(); public int compareTo(ZuulFilter filter) { return Integer.compare(this.filterOrder(), filter.filterOrder()); } } 根据设计图,我们知道Filter有三个阶段 分别是 pre, route,post, 这里的filterType 就是指定filter type(不同type的执行顺序不一样)。filterOrder 方法就是指定相同type 下filter的执行顺序了。同时也是实现 Comparable接口的原因了。 关于自定义 type,我们稍后再看。 那我们来看看 这些type的顺序是怎么指定的, /** * Zuul Servlet filter to run Zuul within a Servlet Filter. The filter invokes pre-routing filters first, * then routing filters, then post routing filters. Handled exceptions in pre-routing and routing * call the error filters, then call post-routing filters. Errors in post-routing only invoke the error filters. * Unhandled exceptions only invoke the error filters * * @author Mikey Cohen * Date: 10/12/11 * Time: 2:54 PM */ public class ZuulServletFilter implements Filter { private ZuulRunner zuulRunner; @Override public void init(FilterConfig filterConfig) throws ServletException { String bufferReqsStr = filterConfig.getInitParameter("buffer-requests"); boolean bufferReqs = bufferReqsStr != null && bufferReqsStr.equals("true") ? true : false; zuulRunner = new ZuulRunner(bufferReqs); } @Override public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException { try { init((HttpServletRequest) servletRequest, (HttpServletResponse) servletResponse); try { preRouting(); } catch (ZuulException e) { error(e); postRouting(); return; } // Only forward onto to the chain if a zuul response is not being sent if (!RequestContext.getCurrentContext().sendZuulResponse()) { filterChain.doFilter(servletRequest, servletResponse); return; } try { routing(); } catch (ZuulException e) { error(e); postRouting(); return; } try { postRouting(); } catch (ZuulException e) { error(e); return; } } catch (Throwable e) { error(new ZuulException(e, 500, "UNCAUGHT_EXCEPTION_FROM_FILTER_" + e.getClass().getName())); } finally { RequestContext.getCurrentContext().unset(); } } } 现在 看到了,ZuulFilter是通过实现在Servlet Filter的生命周期中实现的。那现在我们可以猜想 preRouting() 里会把所有的pre Type 按顺序执行: /** * runs all "pre" filters. These filters are run before routing to the orgin. * * @throws ZuulException */ public void preRoute() throws ZuulException { try { runFilters("pre"); } catch (ZuulException e) { throw e; } catch (Throwable e) { throw new ZuulException(e, 500, "UNCAUGHT_EXCEPTION_IN_PRE_FILTER_" + e.getClass().getName()); } } /** * runs all filters of the filterType sType/ Use this method within filters to run custom filters by type * * @param sType the filterType. * @return * @throws Throwable throws up an arbitrary exception */ public Object runFilters(String sType) throws Throwable { if (RequestContext.getCurrentContext().debugRouting()) { Debug.addRoutingDebug("Invoking {" + sType + "} type filters"); } boolean bResult = false; List<ZuulFilter> list = FilterLoader.getInstance().getFiltersByType(sType); if (list != null) { for (int i = 0; i < list.size(); i++) { ZuulFilter zuulFilter = list.get(i); Object result = processZuulFilter(zuulFilter); if (result != null && result instanceof Boolean) { bResult |= ((Boolean) result); } } } return bResult; } 从代码可以看出process根据列表里的filter顺序执行,顺序在返回前就按order 排序了,这样,当一个 preFilter 执行抛出 ZuulException 之后,pre阶段就结束了,进到最初ServletFilter的后续阶段。 所以到这里为止,我们就知道Zuul是怎么工作和扩展的了,大家就可以根据自己的需要进行扩展了。

优秀的个人博客,低调大师

java源码 - CountDownLatch

开篇 CountDownLatch是一个同步工具类,用来协调多个线程之间的同步,或者说起到线程之间的通信(而不是用作互斥的作用)。 CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后,再继续执行。使用一个计数器进行实现。计数器初始值为线程的数量。当每一个线程完成自己任务后,计数器的值就会减一。当计数器的值为0时,表示所有的线程都已经完成了任务,然后在CountDownLatch上等待的线程就可以恢复执行任务。 CountDownLatch是一次性的,计数器的值只能在构造方法中初始化一次,之后没有任何机制再次对其设置值,当CountDownLatch使用完毕后,它不能再次被使用。 CountDownLatch的用法 CountDownLatch典型用法1:某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为n new CountDownLatch(n) ,每当一个任务线程执行完毕,就将计数器减1 countdownlatch.countDown(),当计数器的值变为0时,在CountDownLatch上 await() 的线程就会被唤醒。一个典型应用场景就是启动一个服务时,主线程需要等待多个组件加载完毕,之后再继续执行。 CountDownLatch典型用法2:实现多个线程开始执行任务的最大并行性。注意是并行性,不是并发,强调的是多个线程在某一时刻同时开始执行。类似于赛跑,将多个线程放到起点,等待发令枪响,然后同时开跑。做法是初始化一个共享的CountDownLatch(1),将其计数器初始化为1,多个线程在开始执行任务前首先 coundownlatch.await(),当主线程调用 countDown() 时,计数器变为0,多个线程同时被唤醒。 CountDownLatch的demo public class CountDownLatchDemo { public static void main(String[] args) throws InterruptedException{ CountDownLatch countDownLatch = new CountDownLatch(2){ @Override public void await() throws InterruptedException { super.await(); System.out.println(Thread.currentThread().getName() + " count down is ok"); } }; Thread thread1 = new Thread(new Runnable() { @Override public void run() { //do something try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " is done"); countDownLatch.countDown(); } }, "thread1"); Thread thread2 = new Thread(new Runnable() { @Override public void run() { try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " is done"); countDownLatch.countDown(); } }, "thread2"); thread1.start(); thread2.start(); countDownLatch.await(); } CountDownLatch的类定义 CountDownLatch内部包含Sync类。 CountDownLatch内部包含Sync类的对象sync。 Sync类继承自AQS(神奇的AQS),构造函数设置AQS的state值为等待值。 public class CountDownLatch { private static final class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 4982264981922014374L; Sync(int count) { setState(count); } int getCount() { return getState(); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } } } private final Sync sync; public CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count); } } CountDownLatch的等待过程 CountDownLatch通过await()进入等待。 CountDownLatch通过await(long timeout, TimeUnit unit)进入超时等待。 public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } CountDownLatch的await()过程 await()通过sync.acquireSharedInterruptibly()获锁。 acquireSharedInterruptibly通过tryAcquireShared()尝试获锁。 tryAcquireShared()判断获锁成功与否的依据是AQS的state的值是否为零。 获锁失败后通过doAcquireSharedInterruptibly()进入锁等待队列CLH。 public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 尝试获锁失败 if (tryAcquireShared(arg) < 0) // doAcquireSharedInterruptibly(arg); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.SHARED); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } CountDownLatch的await(long timeout, TimeUnit unit)过程 await(long timeout, TimeUnit unit)通过sync.tryAcquireSharedNanos()获锁。 tryAcquireSharedNanos()通过doAcquireSharedNanos()尝试获锁。 tryAcquireShared()判断获锁成功与否的依据是AQS的state的值是否为零。 获锁失败后通过doAcquireSharedNanos()进入锁等待队列CLH,和doAcquireSharedInterruptibly()方法相比增加了超时检测机制,通过LockSupport.parkNanos()实现超时。 public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); return tryAcquireShared(arg) >= 0 || doAcquireSharedNanos(arg, nanosTimeout); } private boolean doAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { if (nanosTimeout <= 0L) return false; final long deadline = System.nanoTime() + nanosTimeout; final Node node = addWaiter(Node.SHARED); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return true; } } nanosTimeout = deadline - System.nanoTime(); if (nanosTimeout <= 0L) return false; if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); if (Thread.interrupted()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } } CountDownLatch的唤醒过程 CountDownLatch通过sync.releaseShared(1)释放锁实现state的递减 tryReleaseShared()方法判断锁状态state==0,递减后值为0说明锁已经被释放。 releaseShared()释放锁成功后通过doReleaseShared()方法唤醒所有等待线程。 doReleaseShared()唤醒锁的过程是一个传播性的唤醒,通过线程A唤醒线程B,然后由线程B唤醒线程C的传播性依次唤醒所有等待线程。 public void countDown() { sync.releaseShared(1); } public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; } protected boolean tryReleaseShared(int releases) { for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } } private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; } } 总结 CountDownLatch的工作原理,总结起来就两点(基于AQS实现): 初始化锁状态的值为需要等待的线程数。 判断锁状态是否已经释放,如果锁未释放所有等待锁的线程就会进入等待的CLH队列。 如果锁状态已经释放,那么就会通过传播性唤醒所有的等待线程。

优秀的个人博客,低调大师

java源码 - CyclicBarrier

开篇 CyclicBarrier是一个同步工具类,它允许一组线程互相等待,直到到达某个公共屏障点。与CountDownLatch不同的是该barrier在释放等待线程后可以重用,所以称它为循环(Cyclic)的屏障(Barrier)。 CyclicBarrier支持一个可选的Runnable命令,在一组线程中的最后一个线程到达之后(但在释放所有线程之前),该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态,此屏障操作很有用。 CyclicBarrier的内部实现逻辑基于ReentrantLock实现,可以理解为ReentrantLock的上层应用者,通过ReentrantLock的Condtion实现线程的休眠和唤醒。 CyclicBarrier用法demo public class Test { public static void main(String[] args) { int N = 4; CyclicBarrier barrier = new CyclicBarrier(N); for(int i=0;i<N;i++) new Writer(barrier).start(); } static class Writer extends Thread{ private CyclicBarrier cyclicBarrier; public Writer(CyclicBarrier cyclicBarrier) { this.cyclicBarrier = cyclicBarrier; } @Override public void run() { System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据..."); try { Thread.sleep(5000); //以睡眠来模拟写入数据操作 System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕,等待其他线程写入完毕"); cyclicBarrier.await(); } catch (InterruptedException e) { e.printStackTrace(); }catch(BrokenBarrierException e){ e.printStackTrace(); } System.out.println("所有线程写入完毕,继续处理其他任务..."); } } } 线程Thread-0正在写入数据... 线程Thread-3正在写入数据... 线程Thread-2正在写入数据... 线程Thread-1正在写入数据... 线程Thread-2写入数据完毕,等待其他线程写入完毕 线程Thread-0写入数据完毕,等待其他线程写入完毕 线程Thread-3写入数据完毕,等待其他线程写入完毕 线程Thread-1写入数据完毕,等待其他线程写入完毕 所有线程写入完毕,继续处理其他任务... 所有线程写入完毕,继续处理其他任务... 所有线程写入完毕,继续处理其他任务... 所有线程写入完毕,继续处理其他任务... CyclicBarrier类定义 parties记录一共等待执行个数,count记录依然等待执行的个数。 barrierCommand记录所有待执行的完成后由最后一个线程执行的完成的命令。 Generation的代的概念来实现CyclicBarrier的复用。 构造函数负责初始化parties、count、barrierCommand的核心变量。 public class CyclicBarrier { // 代的类定义 private static class Generation { boolean broken = false; } // 内部通过ReentrantLock实现线程安全的等待 private final ReentrantLock lock = new ReentrantLock(); // 内部通过Lock的condition实现所有waiter的信号通知 private final Condition trip = lock.newCondition(); // 所有等待执行的个数 private final int parties; // 所有等待线程都完成任务后由最后一个线程执行的命令 private final Runnable barrierCommand; // 通过代的概念实现复用 private Generation generation = new Generation(); // 还在等待的个数 private int count; // 核心构造函数 public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; } public CyclicBarrier(int parties) { this(parties, null); } CyclicBarrier工作原理 CyclicBarrier通过ReentrantLock来保证线程休眠和唤醒的通信。 在执行过程中会对等待计数进行减一操作,值不为0当前线程进入休眠等待其他线程唤醒 在执行过程中会对等待计数进行减一操作,值为0当前线程直接执行barrierCommand并且通过nextGeneration方法唤醒其他等待线程 线程的休眠和唤醒都是基于ReentrantLock来实现的 public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } } private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { // 通过lock来保证线程安全 final ReentrantLock lock = this.lock; lock.lock(); try { final Generation g = generation; // 判断generation过期的情况 if (g.broken) throw new BrokenBarrierException(); // 判断线程中断情况 if (Thread.interrupted()) { breakBarrier(); throw new InterruptedException(); } // 递减待执行的个数计数 int index = --count; // 所有待执行任务完成后执行barrierCommand命令 if (index == 0) { // tripped boolean ranAction = false; try { // barrierCommand命令不为null的时候执行该命令 final Runnable command = barrierCommand; if (command != null) command.run(); // 已经执行了barrierCommand ranAction = true; // 重置generation用以复用并且唤醒所有等待的线程 // private void nextGeneration() { // trip.signalAll(); // count = parties; // generation = new Generation(); // } nextGeneration(); return 0; } finally { if (!ranAction) breakBarrier(); } } // 如果count的值不为0,那么当前线程就开始进入等待 // 外层通过lock占用锁,内层通过wait()进入休眠并释放锁 for (;;) { try { if (!timed) // private final Condition trip = lock.newCondition(); trip.await(); else if (nanos > 0L) nanos = trip.awaitNanos(nanos); } catch (InterruptedException ie) { if (g == generation && ! g.broken) { breakBarrier(); throw ie; } else { // We're about to finish waiting even if we had not // been interrupted, so this interrupt is deemed to // "belong" to subsequent execution. Thread.currentThread().interrupt(); } } // 各种后置处理逻辑 if (g.broken) throw new BrokenBarrierException(); if (g != generation) return index; if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } } 参考文章 Java并发编程:CountDownLatch、CyclicBarrier和Semaphore

资源下载

更多资源
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring

Spring框架(Spring Framework)是由Rod Johnson于2002年提出的开源Java企业级应用框架,旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念,提供核心容器、应用上下文、数据访问集成等模块,支持整合Hibernate、Struts等第三方框架,其适用范围不仅限于服务器端开发,绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

用户登录
用户注册