【Java集合】Queue

男娘i 2022-05-18 07:42 214阅读 0赞

Queue用于模拟队列这种数据结构，队列通常是指“先进先出”（FIFO）的容器。新元素插入（offer）到队列的尾部，访问元素（poll）操作会返回队列头部的元素。通常，队列不允许随机访问队列中的元素。

一. Queue的实现

没有实现阻塞接口的LinkedList：实现了java.util.Queue接口和java.util.AbstractQueue接口

(一) 内置的不阻塞队列：PriorityQueue 和 ConcurrentLinkedQueue

PriorityQueue 和 ConcurrentLinkedQueue 类在 Collection Framework 中加入两个具体集合实现。

PriorityQueue 类实质上维护了一个有序列表。加入到 Queue 中的元素根据它们的天然排序（通过其 java.util.Comparable 实现）或者根据传递给构造函数的 java.util.Comparator 实现来定位。

ConcurrentLinkedQueue 是基于链接节点的、线程安全的队列。并发访问不需要同步。因为它在队列的尾部添加元素并从头部删除它们，所以只要不需要知道队列的大 小，

ConcurrentLinkedQueue 对公共集合的共享访问就可以工作得很好。收集关于队列大小的信息会很慢，需要遍历队列。

(二) 实现阻塞接口的：

java.util.concurrent 中加入了 BlockingQueue 接口和五个阻塞队列类。它实质上就是一种带有一点扭曲的 FIFO 数据结构。不是立即从队列中添加或者删除元素，线程执行操作阻塞，直到有空间或者元素可用。

五个队列所提供的各有不同：

1.  ArrayBlockingQueue ：一个由数组支持的有界队列，在构造时需要指定容量， 并可以选择是否需要公平性，如果公平参数被设置true，等待时间最长的线程会优先得到处理（其实就是通过将ReentrantLock设置为true来 达到这种公平性的：即等待时间最长的线程会先操作）。通常，公平性会使你在性能上付出代价，只有在的确非常需要的时候再使用它。它是基于数组的阻塞循环队 列，此队列按 FIFO（先进先出）原则对元素进行排序。
2.  LinkedBlockingQueue ：一个由链接节点支持的可选有界队列，在不指定容量时，容量为Integer.MAX\_VALUE
3.  PriorityBlockingQueue ：一个由优先级堆支持的无界优先级队列，是一个带优先级的 队列，而不是先进先出队列。元素按优先级顺序被移除，该队列也没有上限，在优先阻塞 队列上put时是不会受阻的。虽然此队列逻辑上是无界的，但是由于资源被耗尽，所以试图执行添加操作可能会导致 OutOfMemoryError，但是如果队列为空，那么取元素的操作take就会阻塞，所以它的检索操作take是受阻的。另外，往入该队列中的元 素要具有比较能力。
4.  DelayQueue ：一个由优先级堆支持的、基于时间的调度队列，基于PriorityQueue来实现的）是一个存放Delayed 元素的无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的 Delayed 元素。如果延迟都还没有期满，则队列没有头部，并且poll将返回null。当一个元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于或等于零的值时，则出现期满，poll就以移除这个元素了。此队列不允许使用 null 元素。
5.  SynchronousQueue ：一个利用 BlockingQueue 接口的简单聚集（rendezvous）机制。

二. Queue的操作

1.  add 增加一个元素 如果队列已满，则抛出IIIegaISlabEepeplian异常
2.  remove 移除并返回队列头部的元素 如果队列为空，则抛出NoSuchElementException异常
3.  element 返回队列头部的元素 如果队列为空，则抛出NoSuchElementException异常
4.  offer 添加一个元素并返回true 如果队列已满，则返回false
5.  poll 移除并返问队列头部的元素 如果队列为空，则返回null
6.  peek 返回队列头部的元素 如果队列为空，则返回null
7.  put 添加一个元素 如果队列满，则阻塞
8.  take 移除并返回队列头部的元素 如果队列为空，则阻塞

三. Queue在项目中的应用

在项目中，我们使用了阻塞队列推送消息，并指定初始容量大小为2000. 关键代码：

//定义消息队列接口，并提供简单的操作
    public interface MessageQueue {
    
        public Message get();
    
        public void put(Message message);
    
        public void startPush(PropertiesConfiguration config);
    
        public int messageCount();
    }

//定义一个阻塞队列
    public class BlockingMessageQueue extends Thread implements MessageQueue { 
        private final static BlockingQueue<Message> queue = new LinkedBlockingQueue<Message>(2000);
    
        @Override
        public Message get() {
            try {
                return queue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        }
    
        @Override
        public void put(Message message) {
            try {
                if (message.getProcessInfo() == null && message.getProcessInfoId() != null && message.getProcessInfoId() > 0) {
                    ProcessService processService = (ProcessService) Constants.WEB_APPLICATION_CONTEXT.getBean("processService");
                    message.setProcessInfo(processService.get(message.getProcessInfoId()));
                }
                queue.put(message);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        @Override
        public void startPush(PropertiesConfiguration config) {
            start();
        }
    
        @Override
        public void run() {
            while (true) {
                Message message = this.get();
                if (message != null) {
                    for (MessageSender sender : PushManager.getSenderList()) {
                        try {
       //无论什么异常，都无视
                            sender.send(message);
                        } catch (Exception e) {
                            e.printStackTrace();
                            continue;
                        }
                    }
                }
            }
        }
    
        /* (non-Javadoc) * @see com.ultra.uflow.module.pushMessage.queue.MessageQueue#messageCount() */
        @Override
        public int messageCount() {
            return queue.size();
        }
    
    }

//消息推送入口
    public class PushManager { 
    
        private static Logger logger = Logger.getLogger(PushManager.class);
    
        private final static PushManager pushManager = new PushManager();
    
        private static boolean inited = false;
    
        private static boolean needPush = false;
    
        private static MessageQueue messageQueue;
    
        private static List<MessageSender> senderList = null;
    
        private static String senderNames = "";
    
        public static PushManager getInstance() {
            return pushManager;
        }
    
        public void init() {
            if (inited) {
                return;
            }
    
            PropertiesConfiguration config = null;
            try {
                config = new PropertiesConfiguration("pushMessage.properties");
            } catch (ConfigurationException e) {
                logger.error("pushMessage.properties 推送配置文件不存在，无法推送消息");
                inited = true;
                e.printStackTrace();
            }
    
            if (config.containsKey("needPush")) {
                needPush = config.getBoolean("needPush");
            }
    
            if (!needPush) {
                logger.warn("needPush=false，不需要推送消息");
                return;
            }
    
            try {
                messageQueue = (MessageQueue) Class.forName(config.getString("queue")).newInstance();
            } catch (Exception e) {
                needPush = false;
                logger.error("推送队列不存在，无法推送消息，请检查queue配置");
                e.printStackTrace();
            }
    
            inited = true;
            logger.info("+++++++++++++++消息推送初始化成功+++++++++++");
    
            initSender(config);
    
            messageQueue.startPush(config); //启动一个线程不断轮训队列
    
        }
    
        @SuppressWarnings("unchecked")
        public static void initSender(PropertiesConfiguration config) {
            //初始化sender
            senderList = new ArrayList<MessageSender>();
            List<String> senderNameList = new ArrayList<String>();
    
            @SuppressWarnings("unused")
            List<String> senders = config.getList("sender");
            for (String senderName : senders) {
                String senderClassName = config.getString(senderName + ".class");
                try {
                    MessageSender sender = (MessageSender) Class.forName(senderClassName).newInstance();
                    sender.init(senderName, config);
                    senderList.add(sender);
                    senderNameList.add(senderName);
                } catch (Exception e) {
                    logger.error("sender初始化失败：" + senderName);
                    e.printStackTrace();
                    continue;
                }
            }
            senderNames = StringUtils.join(senders, ",");
    
        }
    
        public static void put(Message message) {
            if (needPush && inited) {
                messageQueue.put(message);
            }
        }
    
        /** * * @param taskInstance * @param processInfoId 非空 * @param description * @param messageType 非空，参看 PushMessageType.java * @param fromUser * @param toUser */
        public static void put(TaskInstance taskInstance, Long processInfoId, String description, Long messageType, String fromUser, String toUser) {
            if (needPush && inited) {
                messageQueue.put(new Message(taskInstance, processInfoId, description, messageType, fromUser, toUser));
            }
        }
    
        public static List<MessageSender> getSenderList() {
            return senderList;
        }
    
        public static String getSenderNames() {
            return senderNames;
        }
    
    }