代码编织梦想

8. 共享模型之工具

8.2 J.U.C

9. BlockingQueue

* BlockingQueue** 原理

LinkedBlockingQueue 原理

1. 基本的入队出队

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    static class Node<E> {
        E item;
        
        /**
         * 下列三种情况之一
         * - 真正的后继节点
         * - 自己, 发生在出队时
         * - null, 表示是没有后继节点, 是最后了
         */
        Node<E> next;
        
        Node(E x) { item = x; }
    }
}

初始化链表 last = head = new Node(null); Dummy (哨兵节点,哑元节点)节点用来占位,item 为 null

image-20230320223706837

当一个节点入队 last = last.next = node;

image-20230320223722340

再来一个节点入队 last = last.next = node;

image-20230320223739789

出队

Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
head = first;
E x = first.item;
first.item = null;
return x;

h = head

image-20230320223812765

first = h.next

image-20230320223826401

h.next = h

image-20230320223839953

head = first

image-20230320223853826

E x = first.item;
first.item = null;
return x;

image-20230320223910826

2. 加锁分析

高明之处在于用了两把锁和 dummy 节点

  • 用一把锁,同一时刻,最多只允许有一个线程(生产者或消费者,二选一)执行

  • 用两把锁,同一时刻,可以允许两个线程同时(一个生产者与一个消费者)执行

    • 消费者与消费者线程仍然串行
    • 生产者与生产者线程仍然串行

线程安全分析

  • 当节点总数大于 2 时(包括 dummy 节点),putLock 保证的是 last 节点的线程安全,takeLock 保证的是head 节点的线程安全。两把锁保证了入队和出队没有竞争

  • 当节点总数等于 2 时(即一个 dummy 节点,一个正常节点)这时候,仍然是两把锁锁两个对象,不会竞争

  • 当节点总数等于 1 时(就一个 dummy 节点)这时 take 线程会被 notEmpty 条件阻塞,有竞争,会阻塞

// 用于 put(阻塞) offer(非阻塞)
private final ReentrantLock putLock = new ReentrantLock();

// 用户 take(阻塞) poll(非阻塞)
private final ReentrantLock takeLock = new ReentrantLock();

put 操作

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    // count 用来维护元素计数
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
        // 满了等待
        while (count.get() == capacity) {
            // 倒过来读就好: 等待 notFull
            notFull.await();
        }
        // 有空位, 入队且计数加一
        enqueue(node);
        c = count.getAndIncrement(); 
        // 除了自己 put 以外, 队列还有空位, 由自己put线程叫醒其他 put 线程
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    // 如果队列中有一个元素, 叫醒 take 线程
    if (c == 0)
        // 这里调用的是 notEmpty.signal() 而不是 notEmpty.signalAll() 是为了减少竞争
        signalNotEmpty();
}

take 操作

public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
        while (count.get() == 0) {
            notEmpty.await();
        }
        x = dequeue();
        c = count.getAndDecrement();
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    // 如果队列中只有一个空位时, 叫醒 put 线程
    // 如果有多个线程进行出队, 第一个线程满足 c == capacity, 但后续线程 c < capacity
    if (c == capacity)
        // 这里调用的是 notFull.signal() 而不是 notFull.signalAll() 是为了减少竞争
        signalNotFull()
        return x;
}

由 put 唤醒 put 是为了避免信号不足

3. 性能比较

主要列举 LinkedBlockingQueue 与 ArrayBlockingQueue 的性能比较

  • Linked 支持有界,Array 强制有界

  • Linked 实现是链表,Array 实现是数组

  • Linked 是懒惰的,而 Array 需要提前初始化 Node 数组

  • Linked 每次入队会生成新 Node,而 Array 的 Node 是提前创建好的

  • Linked 两把锁,Array 一把锁

10. ConcurrentLinkedQueue

ConcurrentLinkedQueue 的设计与 LinkedBlockingQueue 非常像,也是

  • 两把【锁】,同一时刻,可以允许两个线程同时(一个生产者与一个消费者)执行

  • dummy 节点的引入让两把【锁】将来锁住的是不同对象,避免竞争

  • 只是这【锁】使用了 cas 来实现

事实上,ConcurrentLinkedQueue 应用还是非常广泛的

例如之前讲的 Tomcat 的 Connector 结构时,Acceptor 作为生产者向 Poller 消费者传递事件信息时,正是采用了ConcurrentLinkedQueue 将 SocketChannel 给 Poller 使用(高并发场景)

image-20230320235657889

* ConcurrentLinkedQueue 原理

1. 模仿 ConcurrentLinkedQueue

初始代码

package cn.itcast.concurrent.thirdpart.test;
import java.util.Collection;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicReference;
public class Test3 {
    public static void main(String[] args) {
        MyQueue<String> queue = new MyQueue<>();
        queue.offer("1");
        queue.offer("2");
        queue.offer("3");
        System.out.println(queue);
    }
}
class MyQueue<E> implements Queue<E> {
    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        for (Node<E> p = head; p != null; p = p.next.get()) {
            E item = p.item;
            if (item != null) {
                sb.append(item).append("->");
            }
        }
        sb.append("null");
        return sb.toString();
    }
    @Override
    public int size() {
        return 0;
    }
    @Override
    public boolean isEmpty() {
        return false;
    }
    @Override
    public boolean contains(Object o) {
        return false;
    }
    @Override
    public Iterator<E> iterator() {
        return null;
    }
    @Override
    public Object[] toArray() {
        return new Object[0];
    }
    @Override
    public <T> T[] toArray(T[] a) {
        return null;
    }
    @Override
    public boolean add(E e) {
        return false;
    }
    @Override
    public boolean remove(Object o) {
        return false;
    }
    @Override
    public boolean containsAll(Collection<?> c) {
        return false;
    }
    @Override
    public boolean addAll(Collection<? extends E> c) {
        return false;
    }
    @Override
    public boolean removeAll(Collection<?> c) {
        return false;
    }
    @Override
    public boolean retainAll(Collection<?> c) {
        return false;
    }
    @Override
    public void clear() {
    }
    @Override
    public E remove() {
        return null;
    }
    @Override
    public E element() {
        return null;
    }
    @Override
    public E peek() {
        return null;
    }
    public MyQueue() {
        head = last = new Node<>(null, null);
    }
    private volatile Node<E> last;
    private volatile Node<E> head;
    private E dequeue() {
        /*Node<E> h = head;
         Node<E> first = h.next;
         h.next = h;
         head = first;
         E x = first.item;
         first.item = null;
         return x;*/
        return null;
    }
    @Override
    public E poll() {
        return null;
    }
    @Override
    public boolean offer(E e) {
        return true;
    }
    static class Node<E> {
        volatile E item;
        public Node(E item, Node<E> next) {
            this.item = item;
            this.next = new AtomicReference<>(next);
        }
        AtomicReference<Node<E>> next;
    }
}

offer

public boolean offer(E e) {
    Node<E> n = new Node<>(e, null);
    while(true) {
        // 获取尾节点
        AtomicReference<Node<E>> next = last.next;
        // S1: 真正尾节点的 next 是 null, cas 从 null 到新节点
        if(next.compareAndSet(null, n)) {
            // 这时的 last 已经是倒数第二, next 不为空了, 其它线程的 cas 肯定失败
            // S2: 更新 last 为倒数第一的节点
            last = n;
            return true;
        }
    }
}

11. CopyOnWriteArrayList

CopyOnWriteArraySet 是它的马甲

底层实现采用了 写入时拷贝 的思想,增删改操作会将底层数组拷贝一份,更改操作在新数组上执行,这时不影响其它线程的并发读读写分离。 以新增为例:

public boolean add(E e) {
    synchronized (lock) {
        // 获取旧的数组
        Object[] es = getArray();
        int len = es.length;
        // 拷贝新的数组(这里是比较耗时的操作,但不影响其它读线程)
        es = Arrays.copyOf(es, len + 1);
        // 添加新元素
        es[len] = e;
        // 替换旧的数组
        setArray(es);
        return true;
    }
}

这里的源码版本是 Java 11,在 Java 1.8 中使用的是可重入锁而不是 synchronized

其它读操作并未加锁,例如:

public void forEach(Consumer<? super E> action) {
    Objects.requireNonNull(action);
    for (Object x : getArray()) {
        @SuppressWarnings("unchecked") E e = (E) x;
        action.accept(e);
    }
}

适合『读多写少』的应用场景

get 弱一致性

image-20230319173519075

不容易测试,但问题确实存在

迭代器弱一致性

CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>();
list.add(1);
list.add(2);
list.add(3);
Iterator<Integer> iter = list.iterator();
new Thread(() -> {
    list.remove(0);
    System.out.println(list);
}).start();

sleep1s();
while (iter.hasNext()) {
    System.out.println(iter.next());
}

不要觉得弱一致性就不好

  • 数据库的 MVCC 都是弱一致性的表现

  • 并发高和一致性是矛盾的,需要权衡

迭代器弱一致性

CopyOnWriteArrayList<Integer> list = new CopyOnWriteArrayList<>();
list.add(1);
list.add(2);
list.add(3);
Iterator<Integer> iter = list.iterator();
new Thread(() -> {
    list.remove(0);
    System.out.println(list);
}).start();

sleep1s();
while (iter.hasNext()) {
    System.out.println(iter.next());
}

不要觉得弱一致性就不好

  • 数据库的 MVCC 都是弱一致性的表现

  • 并发高和一致性是矛盾的,需要权衡

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/weixin_41405524/article/details/129679533

juc源码——copyonwritearraylist、(array/linked)blockingqueue、concurrentlinkedqueue(1.8)_绕远的偶人的博客-爱代码爱编程

1、CopyOnWriteArrayList public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable { // 用ReentrantLock实现线程安全 final trans

6.JUC-共享模型之工具-爱代码爱编程

共享模型之工具 1 线程池1. 自定义线程池2. ThreadPoolExecutor1) 线程池状态2) 构造方法3) newFixedThreadPool4) newCachedThreadPool5) newSingleThreadExecutor6) 提交任务7) 关闭线程池shutdownshutdownNow其它方法* 模式之 Work

Java多线程并发编程--Java并发包(JUC)-爱代码爱编程

Java多线程并发–Java并发包(JUC) 前言 前一篇文章中,笔者已经介绍了Java多线程的一些基础知识,但是想要成为一名中高级Java程序员还必须懂得Java并发包(JUC)的知识点,而且JUC现在也是面试中必问的知识点了。 1.什么是Java并发包(JUC) 为了更好的支持多线程并发编程,JDK内部提供了大量实用的API。Java并发包(J

Java并发编程基础篇(三)——其他JUC并发工具类的使用方法-爱代码爱编程

Java并发编程基础篇(三)——其他JUC并发工具类的使用方法 除了上一篇中提到的各类锁之外,JUC包也提供了其他可用于并发场景下的同步工具,包括AtomicInteger等原子操作类、CountDownLatch等并发工具类、ConcurrentHashMap等并发集合类。本篇将会重点讲述这类并发工具的概念与使用方法,并简要介绍线程池的使用方法。 1

Java 并发编程下篇 -(JUC、AQS 源码、ReentrantLock 源码)-爱代码爱编程

并发编程已完结,章节如下:Java 并发编程上篇 -(Synchronized 原理、LockSupport 原理、ReentrantLock 原理)Java 并发编程中篇 -(JMM、CAS 原理、Volatile 原理)Java 并发编程下篇 -(线程池)Java 并发编程下篇 -(JUC、AQS 源码、ReentrantLock 源码) 5、J.U

JUC并发编程-爱代码爱编程

JUC并发编程 java.util.concurrentjava.util.concurrent.atomicjava.util.concurrent.locksjava真得可以开启多线程吗? 不可以 public synchronized void start() { /** * This method i

JUC并发编程深入浅出!-爱代码爱编程

一、JUC java.util.concurrent java.util.concurrent.atomic java.util.concurrent.locks 二、进程与线程 进程: 一个程序,qq.exe Music.exe程序的集合 一个进程往往可以包含多个线程,至少包含一个 进程是程序的一次执行过程,是系统运行程序的基本单位,

JUC 并发编程-爱代码爱编程

1 什么是 JUC java.util.concurrent 并发工具包 用来优雅的解决多线程下的高并发问题,JUC 由Doug Lea 设计开发,此乃神人也! 让 Java 程序员膜拜的大神,这个憨态可掬的老者让人又爱又恨! JUC 的基本框架 文章目录 1 什么是 JUC2 线程与进程概念查看进程和线程的方法线程运行原理两种线程模型栈与

JUC_8_共享模型之JUC_线程安全集合类(java并发编程)-爱代码爱编程

线程安全集合类 线程安全集合类概述ConcurrentHashMap练习:单词计数ConcurrentHashMap 原理JDK 7 HashMap 并发死链测试代码死链复现源码分析小结JDK 8 ConcurrentHashMap重要属性和内部类重要方法构造器分析get 流程put 流程size 计算流程JDK 7 ConcurrentHashM

并发编程(七)共享模型之工具(JUC)-爱代码爱编程

一、AQS 原理  1. 概述 全称是 AbstractQueuedSynchronizer,是阻塞式锁和相关的同步器工具的框架。 特点: (1)用 state 属性来表示资源的状态(分独占模式和共享模式),子类需要定义如何维护这个状态,控制如何获取锁和释放锁。 ①getState:获取 state 状态 ②setStat

java并发编程笔记(七)--juc工具类的使用共享模型之不可变-爱代码爱编程

七、juc工具类的使用 1.线程池 2.构造方法 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,

juc并发编程(17.共享模型之不可变)-爱代码爱编程

内容: 不可变类的使用不可变类设计无状态类设计 1.日期转换的问题 问题提出 下面的代码在运行时,由于 SimpleDateFormat 不是线程安全的 SimpleDateFormat sdf = new Sim

jvm监控搭建-爱代码爱编程

文章目录 JVM监控搭建整体架构JolokiaTelegrafInfluxdbGrafana JVM监控搭建 整体架构 JVM 的各种内存信息,会通过 JMX 接口进行暴露。 Jolokia