手把手带你理解java线程池之工作队列workQueue

2024-04-02 19:04:59 588人浏览独家记忆

Python 官方文档：入门教程 => 点击学习

摘要

目录ArrayBlockingQueueSynchronousQueueLinkedBlockingDequeLinkedBlockingQueueDelayDequeLinkedT

线程池之工作队列

ArrayBlockingQueue

采用数组来实现,并采用可重入锁ReentrantLock来做并发控制，无论是添加还是读取，都先要获得锁才能进行操作可看出进行读写操作都使用了ReentrantLock，ArrayBlockingQueue需要为其指定容量


public boolean offer(E e) {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }
    
    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

SynchronousQueue

由于SynchronousQueue源码比较复杂，里面大量的Cas操作，SynchronousQueue没有容器，所以里面是装不了任务的，当一个生产者线程生产一个任务的时候，如果没有对应的消费者消费，那么该生产者会一直阻塞，知道有消费者消费为止。
图示：

如下代码，如果我们将消费者线程注释掉执行，那么生产者哪里将会一直阻塞


package thread.customthreadpool;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;


public class SynchronousQueueTest {

    private static final SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>();

    private static final ExecutorService service = Executors.newCachedThreadPool();

    public static void main(String[] args) {
        
        service.submit(() -> {
            try {
                synchronousQueue.put("liu");
            }catch (Exception e){
                e.printStackTrace();
            }
            System.out.println("Consumer finished spending");
        });

        
        service.submit(() ->{
            try {
                synchronousQueue.take();
            }catch (Exception e){
                e.printStackTrace();
            }
            System.out.println("take over");
        });
    }
}

LinkedBlockingDeque

LinkedBlockingDeque是一个双向队列，底层使用单链表实现，任何一段都可进行元素的读写操作，在初始化LinkedBlockingDeque的时候，我们可以指定容量，也可不指定，如果不指定，则容量为Integer.MAX_VALUE，

注：Deque是双端队列，而Queue是单端队列，双端意思是两端都可以进行读写操作，而单端则只能从一端进，一端出（FIFO）


public LinkedBlockingDeque() {
        this(Integer.MAX_VALUE);
}


package thread.customthreadpool;
import java.util.concurrent.LinkedBlockingDeque;
public class LinkedBlockingDequeTest {

    private static final LinkedBlockingDeque<Integer> deque = new LinkedBlockingDeque<>();

    public static void main(String[] args) throws InterruptedException {
        deque.put(1);
        deque.put(2);
        deque.put(3);
        deque.put(4);
        deque.put(5);
        System.out.println(deque);
        System.out.println("deque size  "+deque.size());
        deque.take();
        deque.take();
        deque.take();
        deque.take();
        deque.take();
        System.out.println(deque);
        System.out.println("deque size  "+deque.size());
    }
}

LinkedBlockingQueue

底层基于单向连表实现，是一个单向队列，具有先进先出(FIFO)特点，使用了ReentrantLock来做并发控制,读写操作都上锁


private final ReentrantLock putLock = new ReentrantLock();
    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;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }
    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();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

DelayDeque

DelayDeque是一个无界队列，添加进DelayDeque的元素会经过compareTo方法计算，然后按照时间进行排序，排在队头的元素是最早到期的，越往后到期时间越长，DelayDeque只能接受Delayed接口类型如图所示，队列里的元素并不是按照先进先出的规则，而是按照过期时间

示例


package thread.customthreadpool.delayDeque;

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class MyDelayed implements Delayed {

    private final String taskName ;
    private final long nowTime = System.currentTimeMillis();
    private final long expireTime ;

    public MyDelayed(String taskName,long expireTime) {
        this.taskName = taskName;
        this.expireTime = expireTime;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert((nowTime+expireTime) - System.currentTimeMillis(),TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        MyDelayed myDelayed = (MyDelayed) o;
        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    }

    @Override
    public String toString() {
        return "MyDelayed{" +
                "taskName='" + taskName + '\'' +
                ", nowTime=" + nowTime +
                ", expireTime=" + expireTime +
                '}';
    }
}


package thread.customthreadpool.delayDeque;

import java.util.concurrent.*;

public class MyDelayQueue {

    private static final DelayQueue<MyDelayed> delayQueue = new DelayQueue<>();

    private static final ExecutorService service = Executors.newCachedThreadPool();

    public static void main(String[] args) throws InterruptedException {
        service.submit(() -> {
            delayQueue.put(new MyDelayed("A-Task",5000));
            delayQueue.put(new MyDelayed("B-Task",4000));
            delayQueue.put(new MyDelayed("C-Task",3000));
            delayQueue.put(new MyDelayed("D-Task",2000));
            delayQueue.put(new MyDelayed("E-Task",1000));
        });
        while (true){
            System.out.println(delayQueue.take());
        }
    }
}

result

应用场景

1.美团外卖订单：当我们下单后没付款，30分钟后将自动取消订单
2.缓存，对于某些任务，需要在特定的时间清理；

and so on

LinkedTransferQueue

当消费线程从队列中取元素时，如果队列为空，那么生成一个为null的节点，消费者线程就一直等待，此时如果生产者线程发现队列中有一个null节点，它就不入队了，而是将元素填充到这个null节点并唤醒消费者线程，然后消费者线程取走元素。
LinkedTransferQueue是 SynchronousQueue 和 LinkedBlockingQueue 的整合，性能比较高，因为没有锁操作， SynchronousQueue不能存储元素，而LinkedTransferQueue能存储元素，

PriorityBlockingQueue

PriorityBlockingQueue是一个无界的阻塞队列，同时是一个支持优先级的队列，读写操作都是基于ReentrantLock, 内部使用堆算法保证每次出队都是优先级最高的元素


public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        E result;
        try {
            while ( (result = dequeue()) == null)
                notEmpty.await();
        } finally {
            lock.unlock();
        }
        return result;
}

到此这篇关于手把手带你理解java线程池之工作队列workQueue的文章就介绍到这了,更多相关java线程池内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

点击免费下载>>软考高级考试备考技巧/历年真题/备考精华资料

--结束END--

本文标题: 手把手带你理解java线程池之工作队列workQueue

本文链接: https://www.lsjlt.com/news/135229.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑，方便收藏和打印～

下载Word文档

去做题

猜你喜欢

手把手带你理解java线程池之工作队列workQueue

目录ArrayBlockingQueueSynchronousQueueLinkedBlockingDequeLinkedBlockingQueueDelayDequeLinkedT...

99+

2024-04-02
学生视角手把手带你写Java 线程池初版

目录Java手写线程池（第一代）手写线程池-定义参数手写线程池-构造器手写线程池-默认构造器手写线程池-execute方法手写线程池-处理任务手写线程池-优雅关闭线程池手写线程池-暴...

99+

2024-04-02
学生视角手把手带你写Java 线程池改良版

目录Java手写线程池（第二代）第二代线程池的优化线程池构造器线程池拒绝策略execute方法手写线程池源码MyExecutorServiceMyRejectedExecutionE...

99+

2024-04-02
手把手带你了解python多进程,多线程

目录多进程多线程线程安全高并发拷贝(多进程，多线程)总结说明相应的学习视频见链接，本文只对重点进行总结。多进程重点（只要看下面代码的main函数即可） 1.创建 2.如何...

99+

2024-04-02
10分钟带你徒手写个Java线程池

目录Java线程池核心原理手撸Java线程池定义核心字段创建内部类WordThread创建ThreadPool类的构造方法创建执行任务的方法完整源码编写测试程序总结Java线程池核心...

99+

2023-05-16

Java手写线程池 Java线程池
java线程池工作队列饱和策略代码示例

线程池（Thread Pool）是并行执行任务收集的实用工具。随着 CPU 引入适合于应用程序并行化的多核体系结构，线程池的作用正日益显现。通过 ThreadPoolExecutor类及其他辅助类，Java 5 引入了这一框架，作为新的并...

99+

2023-05-30

java 线程池队列
Java开发之手把手教你搭建企业级工程SSM框架

目录1.在IDEA界面中创建MavenWeb工程2.在pom.xml中添加如下相关依赖3.web.xml 配置 Spring MVC、Spring4.分别在main目录下创建reso...

99+

2024-04-02
详解Java线程池的使用及工作原理

目录一、什么是线程池？二、线程池要解决什么问题？三、线程池的使用四、常用阻塞队列五、线程工厂六、拒绝策略七、线程池的执行逻辑八、execute()方法九、执行流程一、什么是线程池？ ...

99+

2024-04-02