队列(Queue)

队列(Queue)

队列(Queue)是一种特殊的线性表,特殊之处在于它只允许在表的前端(front)进行删除操作,而在表的后端(rear)进行插入操作,和栈一样,队列是一种操作受限制的线性表。进行插入操作的端称为队尾,进行删除操作的端称为队头。队列中没有元素时,称为空队列。 队列的数据元素又称为队列元素。在队列中插入一个队列元素称为入队,从队列中删除一个队列元素成为出队。因为队列只允许在一段插入,在另一端删除,所以只有最早进入队列的元素才能最先从队列中删除,故队列又称为先进先出(FIFO—first in first out)线性表。 本文图片均来自网络


队列(Queue)和栈(Stack)一样也有链表和数组两种实现。

链表实现

入列代码表示:

public void enqueue(T item) {
       Node oldLast = last;
       last = new Node();
       last.item = item;
       if (isEmpty()) {
           top = last;
       } else {
           oldLast.next = last;
       }
       number++;
   }

出列代码表示:

public T dequeue() {
        T temp = top.item;
        top = top.next;
        number--;
        if (isEmpty())
            last = null;
        return temp;
    }

完整的伪代码如下:

/**
 * Created by zzw on 2017/6/28.
 * Version:
 * Des:
 */

public class MyQuery<T> {

    private Node top;
    private Node last;
    private int number;


    class Node {
        T item;
        Node next;
    }

    public T dequeue() {
        T temp = top.item;
        top = top.next;
        number--;
        if (isEmpty())
            last = null;
        return temp;
    }

    public void enqueue(T item) {
        Node oldLast = last;
        last = new Node();
        last.item = item;
        if (isEmpty()) {
            top = last;
        } else {
            oldLast.next = last;
        }
        number++;
    }


    private boolean isEmpty() {
        return size() == 0;
    }

    private int size() {
        return number;
    }

}

图示如下:


使用数组实现的称为顺序储存,这种方式出队复杂度高并且容易假溢出。 入列

public E enqueue(E item) {
        addElement(item);
        return item;
    }

出列

public E dequeue() {
       if (size() <= front)
           return null;

       E obj = elementAt(front);
       setElementAt(null, front);
       front++;
       return obj;
   }

完整伪代码(继承Vector实现):

/**
 * Created by zzw on 2017/6/28.
 * Version:
 * Des:
 */

public class MyQuery<E> extends Vector<E> {
    int front = 0;

    public E enqueue(E item) {
        addElement(item);
        return item;
    }

    public E dequeue() {
        if (size() <= front)
            return null;

        E obj = elementAt(front);
        setElementAt(null, front);
        front++;
        return obj;
    }
}

如下图所示:


看看队列在Android里面的使用 Handle消息队列 使用Handle的时候都要使用Looper.loop()

 /**
     * Run the message queue in this thread. Be sure to call
     * {@link #quit()} to end the loop.
     */
    public static void loop() {
        final Looper me = myLooper();
        if (me == null) {
            throw new RuntimeException("No Looper; Looper.prepare() wasn't called on this thread.");
        }
        final MessageQueue queue = me.mQueue;

        // Make sure the identity of this thread is that of the local process,
        // and keep track of what that identity token actually is.
        Binder.clearCallingIdentity();
        final long ident = Binder.clearCallingIdentity();

        for (;;) {
            Message msg = queue.next(); // might block
            if (msg == null) {
                // No message indicates that the message queue is quitting.
                return;
            }

            // This must be in a local variable, in case a UI event sets the logger
            final Printer logging = me.mLogging;
            if (logging != null) {
                logging.println(">>>>> Dispatching to " + msg.target + " " +
                        msg.callback + ": " + msg.what);
            }

            final long traceTag = me.mTraceTag;
            if (traceTag != 0 && Trace.isTagEnabled(traceTag)) {
                Trace.traceBegin(traceTag, msg.target.getTraceName(msg));
            }
            try {
                msg.target.dispatchMessage(msg);
            } finally {
                if (traceTag != 0) {
                    Trace.traceEnd(traceTag);
                }
            }

            if (logging != null) {
                logging.println("<<<<< Finished to " + msg.target + " " + msg.callback);
            }

            // Make sure that during the course of dispatching the
            // identity of the thread wasn't corrupted.
            final long newIdent = Binder.clearCallingIdentity();
            if (ident != newIdent) {
                Log.wtf(TAG, "Thread identity changed from 0x"
                        + Long.toHexString(ident) + " to 0x"
                        + Long.toHexString(newIdent) + " while dispatching to "
                        + msg.target.getClass().getName() + " "
                        + msg.callback + " what=" + msg.what);
            }

            msg.recycleUnchecked();
        }
    }

//MessageQueue中
  Message next() {
        // Return here if the message loop has already quit and been disposed.
        // This can happen if the application tries to restart a looper after quit
        // which is not supported.
        final long ptr = mPtr;
        if (ptr == 0) {
            return null;
        }

        int pendingIdleHandlerCount = -1; // -1 only during first iteration
        int nextPollTimeoutMillis = 0;
        for (;;) {
            if (nextPollTimeoutMillis != 0) {
                Binder.flushPendingCommands();
            }

            nativePollOnce(ptr, nextPollTimeoutMillis);

            synchronized (this) {
                // Try to retrieve the next message.  Return if found.
                final long now = SystemClock.uptimeMillis();
                Message prevMsg = null;
                Message msg = mMessages;
                if (msg != null && msg.target == null) {
                    // Stalled by a barrier.  Find the next asynchronous message in the queue.
                    do {
                        prevMsg = msg;
                        msg = msg.next;
                    } while (msg != null && !msg.isAsynchronous());
                }
                if (msg != null) {
                    if (now < msg.when) {
                        // Next message is not ready.  Set a timeout to wake up when it is ready.
                        nextPollTimeoutMillis = (int) Math.min(msg.when - now, Integer.MAX_VALUE);
                    } else {
                        // Got a message.
                        mBlocked = false;
                        if (prevMsg != null) {
                            prevMsg.next = msg.next;
                        } else {
                            mMessages = msg.next;
                        }
                        msg.next = null;
                        if (DEBUG) Log.v(TAG, "Returning message: " + msg);
                        msg.markInUse();
                        return msg;
                    }
                } else {
                    // No more messages.
                    nextPollTimeoutMillis = -1;
                }

                // Process the quit message now that all pending messages have been handled.
                if (mQuitting) {
                    dispose();
                    return null;
                }

                // If first time idle, then get the number of idlers to run.
                // Idle handles only run if the queue is empty or if the first message
                // in the queue (possibly a barrier) is due to be handled in the future.
                if (pendingIdleHandlerCount < 0
                        && (mMessages == null || now < mMessages.when)) {
                    pendingIdleHandlerCount = mIdleHandlers.size();
                }
                if (pendingIdleHandlerCount <= 0) {
                    // No idle handlers to run.  Loop and wait some more.
                    mBlocked = true;
                    continue;
                }

                if (mPendingIdleHandlers == null) {
                    mPendingIdleHandlers = new IdleHandler[Math.max(pendingIdleHandlerCount, 4)];
                }
                mPendingIdleHandlers = mIdleHandlers.toArray(mPendingIdleHandlers);
            }

            // Run the idle handlers.
            // We only ever reach this code block during the first iteration.
            for (int i = 0; i < pendingIdleHandlerCount; i++) {
                final IdleHandler idler = mPendingIdleHandlers[i];
                mPendingIdleHandlers[i] = null; // release the reference to the handler

                boolean keep = false;
                try {
                    keep = idler.queueIdle();
                } catch (Throwable t) {
                    Log.wtf(TAG, "IdleHandler threw exception", t);
                }

                if (!keep) {
                    synchronized (this) {
                        mIdleHandlers.remove(idler);
                    }
                }
            }

            // Reset the idle handler count to 0 so we do not run them again.
            pendingIdleHandlerCount = 0;

            // While calling an idle handler, a new message could have been delivered
            // so go back and look again for a pending message without waiting.
            nextPollTimeoutMillis = 0;
        }
    }

Handle 异步处理中用来存放Message对象的数据结构,按照“先进先出”的原则存放消息。存放并非实际意义的保存,而是将Message对象以链表的方式串联起来的。MessageQueue对象不需要我们自己创建,而是有Looper对象对其进行管理,一个线程最多只可以拥有一个MessageQueue。在Lopper方法中:出现了一个死循环,从队列中不断的取出message,执行msg.target.dispatchMessage(msg);

EventBusEventBus里面中,将消息封装成一个PendingPost

final class PendingPost {
    private final static List<PendingPost> pendingPostPool = new ArrayList<PendingPost>();
    Object event;
    Subscription subscription;
    PendingPost next;
    private PendingPost(Object event, Subscription subscription) {
        this.event = event;
        this.subscription = subscription;
    }
}

在使用队列PendingPostQueue进行管理

final class PendingPostQueue {
    private PendingPost head;
    private PendingPost tail;
    synchronized void enqueue(PendingPost pendingPost) {
        if (pendingPost == null) {
            throw new NullPointerException("null cannot be enqueued");
        }
        if (tail != null) {
            tail.next = pendingPost;
            tail = pendingPost;
        } else if (head == null) {
            head = tail = pendingPost;
        } else {
            throw new IllegalStateException("Head present, but no tail");
        }
        notifyAll();
    }
    synchronized PendingPost poll() {
        PendingPost pendingPost = head;
        if (head != null) {
            head = head.next;
            if (head == null) {
                tail = null;
            }
        }
        return pendingPost;
    }
    synchronized PendingPost poll(int maxMillisToWait) throws InterruptedException {
        if (head == null) {
            wait(maxMillisToWait);
        }
        return poll();
    }
}

Handler发送消息

final class HandlerPoster extends Handler {
    private final PendingPostQueue queue;
    private final int maxMillisInsideHandleMessage;
    private final EventBus eventBus;
    private boolean handlerActive;
    HandlerPoster(EventBus eventBus, Looper looper, int maxMillisInsideHandleMessage) {
        super(looper);
        this.eventBus = eventBus;
        this.maxMillisInsideHandleMessage = maxMillisInsideHandleMessage;
        queue = new PendingPostQueue();
    }
    void enqueue(Subscription subscription, Object event) {
        PendingPost pendingPost = PendingPost.obtainPendingPost(subscription, event);
        synchronized (this) {
            queue.enqueue(pendingPost);
            if (!handlerActive) {
                handlerActive = true;
                if (!sendMessage(obtainMessage())) {
                    throw new EventBusException("Could not send handler message");
                }
            }
        }
    }
    @Override
    public void handleMessage(Message msg) {
        boolean rescheduled = false;
        try {
            long started = SystemClock.uptimeMillis();
            while (true) {
                PendingPost pendingPost = queue.poll();
                if (pendingPost == null) {
                    synchronized (this) {
                        // Check again, this time in synchronized
                        pendingPost = queue.poll();
                        if (pendingPost == null) {
                            handlerActive = false;
                            return;
                        }
                    }
                }
                eventBus.invokeSubscriber(pendingPost);
                long timeInMethod = SystemClock.uptimeMillis() - started;
                if (timeInMethod >= maxMillisInsideHandleMessage) {
                    if (!sendMessage(obtainMessage())) {
                        throw new EventBusException("Could not send handler message");
                    }
                    rescheduled = true;
                    return;
                }
            }
        } finally {
            handlerActive = rescheduled;
        }
    }
}

这里面有个死循环,不断的取出event,通过eventBus.invokeSubscriber(pendingPost);执行相关函数。

相关参考链接: 浅谈算法和数据结构: 一 栈和队列 队列在Android中的使用


水平有限,文中有什么不对或者有什么建议希望大家能够指出,谢谢!

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • [python模块]队列queue

    因为queue使用了一个线程锁(pthread.Lock()),以及三个条件变量(pthread.condition()),来保证了线程安全。

    py3study
  • python队列Queue

    Coxhuang
  • python队列Queue

    Queue是python标准库中的线程安全的队列(FIFO)实现,提供了一个适用于多线程编程的先进先出的数据结构,即队列,用来在生产者和消费者线程之间的信息传递

    周小董
  • C# 队列(Queue)

    //Queue队列就是先进先出。它并没有实现 IList,ICollection。所以它不能按索引访问元素,不能使用Add和Remove。下面是...

    zls365
  • Python| 队列 Queue

    本文算是一次队列的学习笔记,Queue 模块实现了三种类型的队列,它们的区别仅仅是队列中元素被取回的顺序。在 FIFO 队列中,先添加的任务先取回。在 LIFO...

    用户1278550
  • python消息队列Queue

    实例1:消息队列Queue,不要将文件命名为“queue.py”,否则会报异常“ImportError: cannot import name 'Queue'”...

    py3study
  • Python 队列Queue和Prior

    Python的Queue模块适用于多线程编程的FIFO实现。它可用于在生产者(producer)和消费者(consumer)之间线程安全(thread-safe...

    py3study
  • Zookeeper应用之——队列(Queue)

      为了在Zookeeper中实现分布式队列,首先需要设计一个znode来存放数据,这个节点叫做队列节点,我们的例子中这个节点是/zookeeper/queue...

    小忽悠
  • Zookeeper应用之——队列(Queue)

    为了在Zookeeper中实现分布式队列,首先需要设计一个znode来存放数据,这个节点叫做队列节点,我们的例子中这个节点是/zookeeper/queue。

    小忽悠
  • 纸上谈兵: 队列 (queue)

    作者:Vamei 出处:http://www.cnblogs.com/vamei 欢迎转载,也请保留这段声明。谢谢! 队列(queue)是一个简单而常见的数据结...

    Vamei
  • 35.python 线程队列Queue-FIFO

    之前的文章中讲解很多关于线程间通信的知识,比如:线程互斥锁lock,线程事件event,线程条件变量condition 等等,这些都是在开发中经常使用的内容,而...

    猿说编程[Python和C]
  • 数据结构学习之队列(queue)

    songleo
  • Salesforce队列(Queue)应用场景介绍

    Salesforce队列(Queue)帮助你的团队管理销售线索,个案,服务合同以及其它的自定义对象。当你将一条记录手动放入到队列中或者通过个案和线索的自动分配规...

    臭豆腐
  • 05-图解数据结构之队列--Queue

    张风捷特烈
  • Python 进程之间的通信 - 队列Queue

    Process并不能像线程那样共享全局变量,那么它们之间如果有传递值的需求,那么该怎么办呢? 对于这种需求,操作系统提供了很多机制来实现进程间的通信,例如队列...

    Devops海洋的渔夫
  • 数据结构--队列Queue--循环顺序队列

    针对顺序队列中的入队操作:if 队列没满,但是队尾到达数组末尾了,队列"满"了,其实没有满,数据需要整体移至数组头部,才可以继续入队。 为解决该问题,避免数据...

    Michael阿明
  • 数据结构--队列Queue--链式队列、顺序队列

    Michael阿明
  • 数据结构 | Java 队列 —— Queue 详细分析

    摘要: 原创出处 https://www.cnblogs.com/lemon-flm/p/7877898.html 「低调人生」欢迎转载,保留摘要,谢谢!

    芋道源码
  • 记录:简单使用Laravel 队列 queue 教程

    Laravel 队列为不同的后台队列服务提供统一的 API,例如 Beanstalk,Amazon SQS,Redis,甚至其他基于关系型数据库的队列。队列的目...

    Alone88

扫码关注云+社区

领取腾讯云代金券