双缓存队列和异步工具类

矫情吗；* 2023-02-13 14:47 1阅读 0赞

**一、双缓存队列**

import java.util.LinkedList;
    import java.util.Queue;
    
    
    /** * 双缓存队列 * @param <T> */
    public class DoubleBlockingQueue<T> { 
    
        private volatile  boolean direction = true;
    
        private Queue<T> queueOne = new LinkedList<>();
    
        private Queue<T> queueTwo = new LinkedList<>();
    
        private Object readLock = new Object();
    
        private Object writeLock = new Object();
    
        /** * 添加数据 * @param t */
        public void add(T t){ 
            synchronized (readLock){ 
                if(direction){ 
                  queueOne.add(t);
                }else { 
                    queueTwo.add(t);
                }
            }
        }
    
        /** * 获取队列数据 * @return */
        public T poll(){ 
            synchronized (writeLock){ 
                if (direction){ 
                    return unQueue(queueTwo,queueOne);
                }else { 
                    return unQueue(queueOne,queueTwo);
                }
            }
        }
    
    
        private T unQueue(Queue<T> poll, Queue<T> put) { 
            if (poll.isEmpty()) { 
                synchronized (readLock) { 
                    if (put.isEmpty()) { 
                        return null;
                    }
                    direction = !direction;
                }
                return put.remove();
            } else { 
                return poll.remove();
            }
        }
    }

**队列**

import java.util.concurrent.ConcurrentHashMap;
    
    /** * 队列工具 * @param <K> * @param <V> */
    public class OrderedQueuePool<K, V> { 
    
        ConcurrentHashMap<K, TasksQueue<V>> map = new ConcurrentHashMap<>();
    
        /** * 获得任务队列 * * @param key * @return */
        public TasksQueue<V> getTasksQueue(K key) { 
            TasksQueue<V> queue = map.get(key);
            if (queue == null) { 
                TasksQueue<V> newQueue = new TasksQueue<V>();
                queue = map.putIfAbsent(key, newQueue);
                if (queue == null) { 
                    queue = newQueue;
                }
            }
            return queue;
        }
    
        /** * 获得全部任务队列 * * @return */
        public ConcurrentHashMap<K, TasksQueue<V>> getTasksQueues() { 
            return map;
        }
    
        /** * 移除任务队列 * * @param key * @return */
        public void removeTasksQueue(K key) { 
            map.remove(key);
        }
    }

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    /** * Created by jiangwenping on 17/3/10. */
    public class TasksQueue<V> { 
    
        /** * 命令队列 */
        private final BlockingQueue<V> tasksQueue = new LinkedBlockingQueue<>();
    
        private boolean processingCompleted = true;
    
        /** * 下一执行命令 * * @return */
        public V poll() { 
            return tasksQueue.poll();
        }
    
        /** * 增加执行指令 * * @param value * @return */
        public boolean add(V value) { 
            return tasksQueue.add(value);
        }
    
        /** * 清理 */
        public void clear() { 
            tasksQueue.clear();
        }
    
        /** * 获取指令数量 * * @return */
        public int size() { 
            return tasksQueue.size();
        }
    
        public boolean isProcessingCompleted() { 
            return processingCompleted;
        }
    
        public void setProcessingCompleted(boolean processingCompleted) { 
            this.processingCompleted = processingCompleted;
        }
    
    }

**二、异步工具类**

import lombok.extern.slf4j.Slf4j;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.scheduling.annotation.EnableAsync;
    
    import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    @Slf4j
    @EnableAsync
    public class Async { 
    
        static ExecutorService executor = Executors.newFixedThreadPool(1);
    
    
        private void executorAsyncOne(){ 
            CompletableFuture<Void> future = CompletableFuture.runAsync(() ->{ 
                //TODO 执行业务代码
    
            }, executor);
            future.thenAccept((e) -> log.info("数据为:{}",e));
            //关闭线程池
            executor.shutdown();
        }
    
        @Async
        public void executorAsyncTwo(){ 
            //TODO 执行业务代码
    
        }
    }