美文网首页大数据
自定义的线程池实现任务的出入队列

自定义的线程池实现任务的出入队列

作者: 天草二十六_简村人 | 来源:发表于2023-02-20 07:22 被阅读0次

一、抽象类Task

import com.xhtech.hermes.commons.util.UUIDUtil;

import java.util.concurrent.atomic.AtomicBoolean;

public abstract class Task {

    private String id;

    private boolean success = false;

    private AtomicBoolean invalid = new AtomicBoolean(false);

    public abstract void execute();

    public String id() {
        if (id == null) {
            id = UUIDUtil.genId();
        }
        return id;
    }

    public Task expire() {
        return invalid.compareAndSet(false, true) ? this : null;
    }

    public boolean success() {
        return success;
    }

    public Task success(boolean success) {
        this.success = success;
        return this;
    }
}

二、任务队列包装类

import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.task.TaskRejectedException;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class TaskQueueScheduler {

    private final Logger logger = LoggerFactory.getLogger(TaskQueueScheduler.class);

    public static final int DEFAULT_EVENT_LOOP_THREADS = Runtime.getRuntime().availableProcessors() * 2;

    private static final StopTask STOP_TASK = new StopTask();

    private static final String DEFAULT_NAME = TaskQueueScheduler.class.getSimpleName();

    private String name;

    private int queueSize = -1;

    private int poolSize = 0;

    private LinkedBlockingDeque<Task> queue;

    private Map<String, Task> queueing = new ConcurrentHashMap<>();

    private ThreadPoolExecutor executor;

    private LinkedBlockingQueue consumerQueue = new LinkedBlockingQueue();

    private AtomicBoolean running = new AtomicBoolean(false);

    private CountDownLatch stopLatch;

    public TaskQueueScheduler() {
        this(-1, true);
    }

    public TaskQueueScheduler(String name) {
        this(DEFAULT_EVENT_LOOP_THREADS, -1, true, name);
    }

    public TaskQueueScheduler(int poolSize, String name) {
        this(poolSize,-1, true, name);
    }

    public TaskQueueScheduler(int poolSize, int queueSize) {
        this(poolSize, queueSize, true);
    }

    public TaskQueueScheduler(int poolSize, boolean start) {
        this(poolSize, -1, start);
    }

    public TaskQueueScheduler(int poolSize, boolean start, String name) {
        this(poolSize, -1, start, name);
    }

    public TaskQueueScheduler(int poolSize, int queueSize, boolean start) {
        this(queueSize, start, DEFAULT_NAME);
    }

    public TaskQueueScheduler(int poolSize, int queueSize, boolean start, String name) {
        this.poolSize = poolSize;
        this.queueSize = queueSize;
        this.name = name;
        this.queue = queueSize > 0 ? new LinkedBlockingDeque(queueSize) : new LinkedBlockingDeque();
        this.executor = createExecutor(name);

        if (start) {
            start();
        }
    }

    private ThreadPoolExecutor createExecutor(String name) {
        BasicThreadFactory factory = new BasicThreadFactory.Builder().namingPattern(name + "-thread-%d").priority(Thread.MAX_PRIORITY).build();
        return new ThreadPoolExecutor(poolSize, poolSize, 0, TimeUnit.MILLISECONDS, consumerQueue, factory);
    }

    public synchronized void add(Task task) throws TaskRejectedException {
        if (queueSize != -1 && queue.size() > queueSize) {
            throw new TaskRejectedException("The queue is full");
        }

        if (task == STOP_TASK || enqueueing(task)) {
            queue.add(task);
        } else {
            logger.info("Task processing in queue");
        }
    }

    public synchronized void addFirst(Task task) throws TaskRejectedException {
        if (queueSize != -1 && queue.size() > queueSize) {
            throw new TaskRejectedException("The queue is full");
        }

        if (enqueueing(task)) {
            queue.addFirst(task);
        } else {
            logger.info("Task processing in queue");
        }
    }

    private boolean enqueueing(Task task) {
        return queueing.putIfAbsent(task.id(), task) == null;
    }

    private boolean dequeueing(Task task) {
        return queueing.remove(task.id()) != null;
    }

    private Task dequeueing(String id) {
        return queueing.remove(id);
    }

    public Task getQueueingTask(String id) {
        Task task = dequeueing(id);
        return task != null ? task.expire() : null;
    }

    public int size() {
        return queue.size();
    }

    public void start() {
        if (running.compareAndSet(false, true)) {
            stopLatch = new CountDownLatch(poolSize);

            for (int i = 0; i < poolSize; i++) {
                executor.submit(new TaskQueueConsumer());
            }
        }
    }

    public void stop() {
        logger.info("try stop...");
        if (running.compareAndSet(true, false)) {
            logger.info("it's in running");
            for (int i = 0; i < poolSize; i++) {
                add(STOP_TASK);
            }

            try {
                logger.info("start await");
                stopLatch.await();
                logger.info("end await");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        logger.info("finish stop...");
    }

    public void shutdown() {
        stop();
        executor.shutdown();
    }

    public boolean isRunning() {
        return running.get();
    }

    class TaskQueueConsumer implements Runnable {
        @Override
        public void run() {
            try {
                Task task;
                while (running.get() && (task = queue.take()) != null && task != STOP_TASK) {
                    if (dequeueing(task) && task.expire() != null) {
                        try {
                            task.execute();
                        } catch (Throwable e) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            } catch (Throwable e) {
                logger.error(e.getMessage(), e);
            } finally {
                logger.warn("{}[{}] is shutdown", name, Thread.currentThread().getName());
                stopLatch.countDown();
            }
        }
    }

    static class StopTask extends Task {
        @Override
        public void execute() {
        }
    }
}

三、测试

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

@Slf4j
public class TaskQueueSchedulerTest {

    private TaskQueueScheduler taskQueueScheduler;
    public void addTask(Task task) {
        if (taskQueueScheduler == null) {
            synchronized (this) {
                if (taskQueueScheduler == null) {
                    taskQueueScheduler = new TaskQueueScheduler(4, getClass().getSimpleName());
                }
            }
        }

        taskQueueScheduler.add(task);
    }


    @Test
    public void test(){
        taskQueueScheduler = new TaskQueueScheduler(4, getClass().getSimpleName());
        taskQueueScheduler.start();
        for (int i = 0; i < 10; i++) {
            final int k = i;
            taskQueueScheduler.add(new Task() {
                @Override
                public void execute() {
                    doSth(2 + k);
                }
            });
        }

        log.info("start stop");

        taskQueueScheduler.stop();
        log.info("end stop");


    }

    private void doSth(long ms){
        try {

            log.info("start {}", ms);
            Thread.sleep(ms);
            log.info("end {}", ms);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

相关文章

  • 并发--共享模型之工具

    线程池 1.1 自定义线程池 先自定义任务队列 自定义线程池 测试: 定义拒绝策略接口: 1.2 ThreadPo...

  • Thread

    队列 线程锁 多线程,线程池 队列 多线程爬虫示例 多线程 自定义线程 线程池

  • 自定义的线程池实现任务的出入队列

    一、抽象类Task 二、任务队列包装类 三、测试

  • 线程池工具

    功能简介 固定线程、限制最大队列长度的自定义线程池; 定制线程池加载任务、子线程各种参数,如分页大小、是否子线程出...

  • C++ 实现线程池

    c++实现一个线程池,首先需要有两个数据结构,一个是线程池,一个是任务队列。为了每个线程线程安全的从任务队列里面拿...

  • C++11 ThreadPool的应用

    线程池的应用 代码结构 任务队列 线程池 工作线程 代码如下

  • 线程池的工作过程

    线程池的工作过程 线程池刚创建时,里面没有一个线程。任务队列是作为参数传进来的。不过,就算队列里面有任务,线程池也...

  • 自己实现一个简单的线程池

    线程池是开发中常用的工具,要想掌握线程池,最好的方法就是自己手动实现一个。 任务类 线程池类 关于队列的选择 之所...

  • 10:Java中的线程池

    1:java中的线程池实现原理 2:上图中的任务队列(用于保存等待执行的任务的阻塞队列)介绍 (1)ArrayBl...

  • 简易线程池的实现

    构成线程池的基本元素 线程池中的线程 任务队列 生产者 消费者 线程池 消费者 生产者 问题 任务队列的大小:如果...

网友评论

    本文标题:自定义的线程池实现任务的出入队列

    本文链接:https://www.haomeiwen.com/subject/myoikdtx.html