队列

队列是一种罕用的数据结构，它最大的特点是先入先出，即先进入队列中的元素最先进去。 滑动窗口的平均值请实现如下类型MovingAverage，计算滑动窗口中所有数字的平均值，该类型构造函数的参数确定滑动窗口的大小，每次调用成员函数next时都会在滑动窗口中增加一个整数，并返回滑动窗口中所有数字的平均值。/** * Initialize your data structure here. * @param {number} size */var MovingAverage = function(size) { this.size = size; this.arr = [] this.sum = 0};/** * @param {number} val * @return {number} */MovingAverage.prototype.next = function(val) { this.arr.push(val) this.sum += val if(this.arr.length>this.size) { this.sum -= this.arr.shift() } return this.sum / this.arr.length};最近申请次数请实现如下类型RecentCounter，它是统计过来3000ms内的申请次数的计数器。该类型的构造函数RecentCounter初始化计数器，申请数初始化为0；函数ping（int t）在工夫t增加一个新申请（t示意以毫秒为单位的工夫），并返回过来3000ms内（工夫范畴为[t-3000，t]）产生的所有申请数。假如每次调用函数ping的参数t都比之前调用的参数值大。var RecentCounter = function() { this.arr = []};/** * @param {number} t * @return {number} */RecentCounter.prototype.ping = function(t) { this.arr.push(t) while(this.arr.length > 1 && this.arr[0] + 3000 < t) { this.arr.shift() } return this.arr.length};二叉树的广度优先遍历应聘者在面试时常常须要应用队列来解决与广度优先搜寻相干的问题。图的广度优先遍历请看图的章节。 ...

摘要：本文通过剖析鸿蒙轻内核队列模块的源码，把握队列应用上的差别。本文分享自华为云社区《鸿蒙轻内核M核源码剖析系列十三 音讯队列Queue》，作者：zhushy 。 队列（Queue）是一种罕用于工作间通信的数据结构。工作可能从队列外面读取音讯，当队列中的音讯为空时，挂起读取工作；当队列中有新音讯时，挂起的读取工作被唤醒并解决新音讯。工作也可能往队列里写入音讯，当队列曾经写满音讯时，挂起写入工作；当队列中有闲暇音讯节点时，挂起的写入工作被唤醒并写入音讯。如果将读队列和写队列的超时工夫设置为0，则不会挂起工作，接口会间接返回，这就是非阻塞模式。音讯队列提供了异步解决机制，容许将一个音讯放入队列，但不立刻解决。同时队列还有缓冲音讯的作用。 本文通过剖析鸿蒙轻内核队列模块的源码，把握队列应用上的差别。本文中所波及的源码，以OpenHarmony LiteOS-M内核为例，均能够在开源站点https://gitee.com/openharmony... 获取。 接下来，咱们看下队列的构造体，队列初始化，队列罕用操作的源代码。 1、队列构造体定义和罕用宏定义1.1 队列构造体定义在文件kernel\include\los_queue.h中定义队列管制块构造体为LosQueueCB，构造体源代码如下。队列状态.queueState取值OS_QUEUE_UNUSED、OS_QUEUE_INUSED，其余构造体成员见正文局部。 typedef struct { UINT8 *queue; /**< 队列内存空间的指针 */ UINT16 queueState; /**< 队列的应用状态 */ UINT16 queueLen; /**< 队列长度，即音讯数量 */ UINT16 queueSize; /**< 音讯节点大小 */ UINT16 queueID; /**< 队列编号 */ UINT16 queueHead; /**< 音讯头节点地位 */ UINT16 queueTail; /**< 音讯尾节点地位 */ UINT16 readWriteableCnt[OS_READWRITE_LEN]; /**< 2维数组，可读、可写的音讯数量, 0:可读, 1:可写 */ LOS_DL_LIST readWriteList[OS_READWRITE_LEN]; /**< 2维双向链表数组，阻塞读、写工作的双向链表, 0:读链表, 1:写链表 */ LOS_DL_LIST memList; /**< 内存节点双向链表 */} LosQueueCB;1.2 队列罕用宏定义零碎反对创立多少队列是依据开发板状况应用宏LOSCFG_BASE_IPC_QUEUE_LIMIT定义的，每一个队列queueID是queueID类型的，取值为[0,LOSCFG_BASE_IPC_QUEUE_LIMIT)，示意队列池中各个队列的编号。 ⑴处的宏从队列池中获取指定队列编号QueueID对应的队列管制块。⑵处依据双向链表节点readWriteList[OS_QUEUE_WRITE]获取队列管制块内存地址。 ...

摘要：对于队列来说数据结构相比栈简单一些，然而也不是很难，搞懂先进先出而后用数组或者链表实现即可。本文分享自华为云社区《手写各种队列，一文搞定》，原文作者：bigsai 。 前言栈和队列是一对好兄弟，栈的机制绝对简略，后入先出，就像进入一个狭小的山洞，山洞只有一个出入口，只能后进先出(在里面的先进来，堵在外面先进去的就有点晦气)。而队列就好比是一个隧道，前面的人跟着后面走，后面人先进来(先入先出)。日常的排队就是队列运行模式的一个形容！ 栈是一种喜新厌旧的数据结构，来了新的就会解决新的把老的先停滞在这(咱们找人、约人办事最厌恶这种人)，队列就是假公济私的一种数据结构，排队先来先得，考究程序性，所以这种数据结构在程序设计、中间件等都十分宽泛的利用，例如音讯队列、FIFO磁盘调度、二叉树层序遍历、BFS宽度优先搜寻等等。 队列的核心理念就是：先进先出！ 队列的概念：队列是一种非凡的线性表，非凡之处在于它只容许在表的前端（front）进行删除操作，而在表的后端（rear）进行插入操作，和栈一样，队列是一种操作受限制的线性表。进行插入操作的端称为队尾，进行删除操作的端称为队头。 队列介绍咱们设计队列时候能够抉择一个规范，这里就拿力扣622设计循环队列作为队列设计的规范。 队头front：删除数据的一端。 队尾rear ：插入数据的一端。 对于数组，从数组前面插入更容易，数组后面插入较艰难，所以个别用数组实现的队列队头在数组后面，队尾在数组前面；而对于链表，插入删除在中间别离进行那么头部(后面)删除尾部插入最不便的抉择。 实现办法： MyCircularQueue(k): 结构器，设置队列长度为 k 。Front: 从队首获取元素。如果队列为空，返回 -1 。Rear: 获取队尾元素。如果队列为空，返回 -1 。enQueue(value): 向循环队列插入一个元素。如果胜利插入则返回真。deQueue(): 从循环队列中删除一个元素。如果胜利删除则返回真。isEmpty(): 查看循环队列是否为空。isFull(): 查看循环队列是否已满。一般队列依照上述的介绍，咱们很容易晓得数组实现的形式。用数组模仿示意队列。要思考初始化，插入，问题。 在这个一般队列一些操作须要留神的有： 初始化：数组的front和rear都指向0. (front和rear下标相等的时候阐明队列为空) 入队：队不满，数组不越界，先队尾地位传值，再队尾下标+1(队尾rear实际上超前一位，为了辨别空队列状况) 出队：队不空，先取队头地位元素，在队头+1。 然而很容易发现问题，每个空间域只能利用一次，造成空间极度节约，非常容易越界！ 循环队列(数组实现)针对上述的问题。有个较好的解决办法！就是对曾经申请的(数组)内存反复利用。这就是咱们所说的循环队列。循环队列的一个益处是咱们能够利用这个队列之前用过的空间。在一个一般队列里，一旦一个队列满了，咱们就不能插入下一个元素，即便在队列后面仍有空间。然而应用循环队列，咱们能应用这些空间去存储新的值。 数组实现的循环队列就是在逻辑上作批改。因为咱们队列中只须要front和rear两个指针。rear在逻辑上在前面，front在逻辑上是在后面的，但实际上它们不肯定谁在前谁在后，在计算间隔的时候须要给rear先补上数组长度减去front，而后求余即可。 初始化：数组的front和rear都指向0. 这里须要留神的是：front和rear位于同一个地位时候，证实队列外面是空的。还有在这里我具体实现时候将数组申请大了一个地位空进去，避免队列满的状况又造成front和rear在同一个地位。 入队：队不满，先队尾地位传值，再rear=(rear + 1) % maxsize; 出队：队不空，先取队头地位元素，front=(front + 1)% maxsize; 这里出队入队指标相加如果遇到最初须要转到头地位，这里间接+1求余找到地位(相比判断是否在最初更加简洁)，其中maxsize是数组理论大小。 是否为空：return rear == front; 大小：return (rear+maxsize-front)%maxsize; 这里很容易了解，一张图就能解释分明，无论是front理论在前在后都能满足要求。 这外面有几个大家须要留神的，就是指标相加如果遇到最初须要转到头的话。能够判断是否到数组开端地位。也能够间接+1求余。其中maxsize是数组理论大小。 具体实现： public class MyCircularQueue { private int data[];// 数组容器 private int front;// 头 private int rear;// 尾 private int maxsize;// 最大长度 public MyCircularQueue(int k) { data = new int[k+1]; front = 0; rear = 0; maxsize = k+1; } public boolean enQueue(int value) { if (isFull()) return false; else { data[rear] = value; rear=(rear + 1) % maxsize; } return true; } public boolean deQueue() { if (isEmpty()) return false; else { front=(front+1)%maxsize; } return true; } public int Front() { if(isEmpty()) return -1; return data[front]; } public int Rear() { if(isEmpty()) return -1; return data[(rear-1+maxsize)%maxsize]; } public boolean isEmpty() { return rear == front; } public boolean isFull() { return (rear + 1) % maxsize == front; }}循环队列(链表实现)对于链表实现的队列，要依据先进先出的规定思考头和尾的地位 ...

明天来看下LeetCode第239题-滑动窗口最大值。 首先看下题： 给你一个整数数组 nums，有一个大小为 k 的滑动窗口从数组的最左侧挪动到数组的最右侧。你只能够看到在滑动窗口内的 k 个数字。滑动窗口每次只向右挪动一位。 返回滑动窗口中的最大值。 输出：nums = [1,3,-1,-3,5,3,6,7], k = 3输入：[3,3,5,5,6,7]解释：滑动窗口的地位 最大值 [1 3 -1] -3 5 3 6 7 31 [3 -1 -3] 5 3 6 7 31 3 [-1 -3 5] 3 6 7 51 3 -1 [-3 5 3] 6 7 51 3 -1 -3 [5 3 6] 7 61 3 -1 -3 5 [3 6 7] 7 就是给出一个数组和一个k值，在数组里会有一个从0开始遍历到n-k+1地位的窗口，要给出每个窗口的最大值。 这题的解法有很多，咱们先来大略理下思路： 1、暴力 O(n*k) 2、堆 O(n*logk) ...

这篇文章是对LeetCode641题设计循环双端队列思路进行记录。(PS:后续应该还会在写一篇，等我把双向链表实现搞清楚了，一个指针还能理清，两个指针pre和next看着就有点晕了，等我消化下。) 先看下题： 设计实现双端队列。你的实现须要反对以下操作： MyCircularDeque(k)：构造函数,双端队列的大小为k。insertFront()：将一个元素增加到双端队列头部。 如果操作胜利返回 true。insertLast()：将一个元素增加到双端队列尾部。如果操作胜利返回 true。deleteFront()：从双端队列头部删除一个元素。 如果操作胜利返回 true。deleteLast()：从双端队列尾部删除一个元素。如果操作胜利返回 true。getFront()：从双端队列头部取得一个元素。如果双端队列为空，返回 -1。getRear()：取得双端队列的最初一个元素。 如果双端队列为空，返回 -1。isEmpty()：查看双端队列是否为空。isFull()：查看双端队列是否满了。示例： MyCircularDeque circularDeque = new MycircularDeque(3); // 设置容量大小为3circularDeque.insertLast(1); // 返回 truecircularDeque.insertLast(2); // 返回 truecircularDeque.insertFront(3); // 返回 truecircularDeque.insertFront(4); // 曾经满了，返回 falsecircularDeque.getRear(); // 返回 2circularDeque.isFull(); // 返回 truecircularDeque.deleteLast(); // 返回 truecircularDeque.insertFront(4); // 返回 truecircularDeque.getFront(); // 返回 4 家喻户晓栈和队列都是限制性的数据结构，在某些业务场景常常用到，而栈和队列其实用两种根本的数据结构数组和链表都能实现的。明天咱们先来看看如何用数组设计循环双端队列。 class MyCircularDeque { private int[] items; private int count; private int head; private int tail; public MyCircularDeque(int k) { this.count = k+1; this.items = new int[count]; this.head = 0; this.tail = 0; } public boolean insertFront(int value) { if (isFull()) { return false; } head = (head - 1 + count) % count; items[head] = value; return true; } public boolean insertLast(int value) { if (isFull()) { return false; } items[tail] = value; tail = (tail + 1) % count; return true; } public boolean deleteFront() { if (isEmpty()) { return false; } head = (head + 1) % count; return true; } public boolean deleteLast() { if (isEmpty()) { return false; } tail = (tail - 1 + count) % count; return true; } public int getFront() { if (isEmpty()) { return -1; } return items[head]; } public int getRear() { if (isEmpty()) { return -1; } return items[(tail - 1 + count) % count]; } public boolean isEmpty() { return head == tail; } public boolean isFull() { return (tail + 1) % count == head; }}首先循环队列你能够设想成一个环状，首尾是相连的，再来它的元素地位是随插入删除始终挪动的，因而效率其实是偏低的，插入元素会波及数据迁徙，所以工业上的队列个别是用链表实现的。 ...

摘要：本文通过剖析LiteOS队列模块的源码，把握队列应用上的差别。本文分享自华为云社区《LiteOS内核源码剖析系列十 音讯队列Queue》，原文作者：zhushy 。 队列（Queue）是一种罕用于工作间通信的数据结构。工作可能从队列外面读取音讯，当队列中的音讯为空时，挂起读取工作；当队列中有新音讯时，挂起的读取工作被唤醒并解决新音讯。工作也可能往队列里写入音讯，当队列曾经写满音讯时，挂起写入工作；当队列中有闲暇音讯节点时，挂起的写入工作被唤醒并写入音讯。如果将读队列和写队列的超时工夫设置为0，则不会挂起工作，接口会间接返回，这就是非阻塞模式。音讯队列提供了异步解决机制，容许将一个音讯放入队列，但不立刻解决。同时队列还有缓冲音讯的作用。 本文通过剖析LiteOS队列模块的源码，把握队列应用上的差别。LiteOS队列模块的源代码，均能够在LiteOS开源站点https://gitee.com/LiteOS/LiteOS 获取。队列源代码、开发文档，示例程序代码如下： LiteOS内核队列源代码包含队列的公有头文件kernel\base\include\los_queue_pri.h、头文件kernel\include\los_queue.h、C源代码文件kernel\base\los_queue.c。 开发指南文档–队列在线文档https://gitee.com/LiteOS/Lite... 接下来，咱们看下队列的构造体，队列初始化，队列罕用操作的源代码。 1、队列构造体定义和罕用宏定义1.1 队列构造体定义在文件kernel\base\include\los_queue_pri.h定义的队列管制块构造体为LosQueueCB，构造体源代码如下。队列状态.queueState取值OS_QUEUE_UNUSED、OS_QUEUE_INUSED，队列内存类型.queueMemType取值OS_QUEUE_ALLOC_STATIC、OS_QUEUE_ALLOC_DYNAMIC，别离示意计用户调配队列内存空间或零碎主动调配。 typedef struct { UINT8 *queueHandle; /**< 队列内存空间的指针 */ UINT8 queueState; /**< 队列的应用状态 */ UINT8 queueMemType; /**< 队列内存类型，用户调配队列内存空间或零碎主动调配 */ UINT16 queueLen; /**< 队列长度，音讯数量 */ UINT16 queueSize; /**< 音讯节点大小 */ UINT32 queueId; /**< 队列编号 */ UINT16 queueHead; /**< 音讯头节点地位 */ UINT16 queueTail; /**< 音讯尾节点地位 */ UINT16 readWriteableCnt[OS_QUEUE_N_RW]; /**< 2维数组，可读、可写的音讯数量, 0:可读, 1:可写 */ LOS_DL_LIST readWriteList[OS_QUEUE_N_RW]; /**< 2维链表数组，阻塞读、写工作的双向链表, 0:读链表, 1:写链表 */ LOS_DL_LIST memList; /**< 内存节点双向链表，兼容CMSIS时应用 */} LosQueueCB;1.2 队列罕用宏定义零碎反对创立多少队列是依据开发板状况应用宏LOSCFG_BASE_IPC_QUEUE_LIMIT定义的，每一个队列queueId是UINT32类型的。队列queueId由2局部组成：count和index，别离处于高16位和低16位。创立队列，应用后删除时，队列回收到队列池时，队列queueId的高16位即count值会加1，这样能够用来示意该队列被创立删除的次数。index取值为[0,LOSCFG_BASE_IPC_QUEUE_LIMIT)，示意队列池中各个队列的编号。 ...

前言接入微信支付的时候，看到微信支付的回调是按照某种频率去回调的，像15s/15s/30s/3m/10m/20m/30m/30m/30m/60m/3h/3h/3h/6h/6h这样，其中有一次成功就不会再回调。于是在想怎么用Java做这个事情。有定时任务这类功能的框架像Spring和Quartz貌似都没有直接提供以上的功能。也是出于想练手的目的，决定自己写一写。 最终的实现效果// 具体的业务BaseJob task = new BaseJob() { // 任务执行的次数（模拟真实业务上的退出） int runTime = 1; @Override public void run() { // 业务逻辑 System.out.println("hello world"); // 这里模拟了微信回调成功，任务完成 if (runTime++ > 3) { this.setExit(true); } }};/** * 测试按照指定时间隔执行某个任务 * @throws IOException */@Testpublic void test1() throws IOException { // 新建一个产生指定时间的延迟时间生成器，内部就是个队列 DesignatDTGenerator designatDTGenerator = new DesignatDTGenerator(); // 设置时间间隔 designatDTGenerator.addDelayTime(1_000) // 1秒后执行 .addDelayTime(4_000) // 距离上次执行4秒后执行 .addDelayTime(15_000) // 距离上次执行15秒后执行 .addDelayTime(180_000) // 距离上次执行3分钟后执行 .addDelayTime(180_000) // 距离上次执行3分钟后执行 .addDelayTime(360_000) // 距离上次执行6分钟后执行 .addDelayTime(3_600_000); // 距离上次执行1小时后执行 // 构造一个提交的任务，传入具体的业务对象task，传入延迟时间生成器designatDTGenerator DelayTimeJob delayTimeJob = new DelayTimeJob(task, designatDTGenerator); // 新建一个执行器，执行器可以重复使用，每次提交新的任务即可 JobActuator actuator = new JobActuator(); // 提交任务，开始执行任务 actuator.addJob(delayTimeJob); // 阻塞主线程，方便查看运行结果 System.in.read();}/** * 测试按照固定时间间隔执行某个任务 * 只是延迟时间生成器不同而已，可以达到不同的调用效果 * @throws IOException */@Testpublic void test2() throws IOException { // 新建一个执行器 JobActuator actuator = new JobActuator(); // 新建一个产生固定时间的延迟时间生成器，每3s执行一次 FixedRateDTGenerator fixedRateDTGenerator = new FixedRateDTGenerator(3000); // 新建一个任务 DelayTimeJob delayTimeJob = new DelayTimeJob(task, fixedRateDTGenerator); // 提交任务，开始执行任务 actuator.addJob(delayTimeJob); // 阻塞主线程，方便查看运行结果 System.in.read();}类图 ...

栈的定义栈是计算机科学中的一种抽象数据类型，只允许在有序的线性数据集合的一端（称为堆栈顶端，英语：top）进行加入数据（英语：push）和移除数据（英语：pop）的运算。因而按照后进先出（LIFO, Last In First Out）的原理运作。（百科全书） 栈的常用操作<font size="3">栈中有两个基本的操作 推入 ：从栈的顶端推入一个数据，依次往下推弹出 ：讲栈顶端的数据移除栈的基本提点就是 先进后出，后进先出除了头尾的节点，每个元素都有一个先驱和一个后继对于栈的画面的理解，可以想象成一个步枪弹夹添加子弹和射击的过程弹夹只有一个出入口进行推入和弹出</font> js模拟实现一个栈<body style='width: 80%;height:80%; margin: 10% auto;'><ul id="stackBox" style="width:100px;border:1px solid #1fb19e;"></ul><div style="width:400px; display:flex;"> <input id="funStackBox" value="执行函数"> <div style="margin-left:100px;"> <button onclick="pushStack()" type='primary'>进栈</button> <button onclick="popStack()" type='primary'>出栈</button> </div></div><script> // 栈盒子 let stackBox = document.getElementById('stackBox') // 执行函数盒子 let funStackBox = document.getElementById('funStackBox') // 栈类 class Stack { constructor(length) { this.list = [] this.length = length } // 新增栈 addStack(val) { if (this.length > this.list.length) { this.list.push(val) let beforeDom = document.getElementById(this.list.length - 1) // 新增栈 let el = document.createElement('li') el.id = this.list.length el.innerText = this.list[this.list.length - 1] stackBox.insertBefore(el, beforeDom) }else{ console.log('栈满了') } } // 弹出栈 popStack() { let delDom = document.getElementById(this.list.length) stackBox.removeChild(delDom) return this.list.pop() } // 栈的大小 stackLength() { console.log('当前栈大小为'+this.list.length) return this.list.length } // 栈的顶层元素 stackIsHas() { return this.list.length?this.list[this.list.length]:console.log('没有栈了') } //查看所有元素 showStack() { return this.list.join('\n'); } //清空队列 clearStack() { this.list = []; this.showStack() console.log('栈空了') } } stackOne = new Stack(5) /** * @author: 周靖松 * @information: 进栈 * @Date: 2019-06-10 12:47:16 */ function pushStack() { stackOne.addStack(funStackBox.value) console.log(funStackBox.value, '进栈') stackOne.stackLength() stackOne.stackIsHas() } /** * @author: 周靖松 * @information: 出栈 * @Date: 2019-06-10 12:47:16 */ function popStack() { let popStack = stackOne.popStack(funStackBox.value) console.log(popStack, '出栈') stackOne.stackLength() stackOne.stackIsHas() }</script></body><font size="3">效果图如下</font> ...

项目地址Github: https://github.com/wuzhc/gmq 1. 概述gmq是基于redis提供的特性,使用go语言开发的一个简单易用的队列;关于redis使用特性可以参考之前本人写过一篇很简陋的文章Redis 实现队列; gmq的灵感和设计是基于有赞延迟队列设计,文章内容清晰而且很好理解,但是没有提供源码,在文章的最后也提到了一些未来架构方向; gmq不是简单按照有赞延迟队列的设计实现功能,在它的基础上,做了一些修改和优化,主要如下: 功能上 多种任务模式,不单单只是延迟队列;例如:延迟队列,普通队列,优先级队列架构上: 添加job由dispatcher调度分配各个bucket,而不是由timer每个bucket维护一个timer,而不是所有bucket一个timertimer每次扫描bucket到期job时,会一次性返回多个到期job,而不是每次只返回一个jobtimer的扫描时钟由bucket中下个job到期时间决定,而不是每秒扫描一次2. 应用场景延迟任务 延迟任务,例如用户下订单一直处于未支付状态，半个小时候自动关闭订单异步任务 异步任务,一般用于耗时操作,例如群发邮件等批量操作超时任务 规定时间内(TTR)没有执行完毕或程序被意外中断,则消息重新回到队列再次被消费,一般用于数据比较敏感,不容丢失的优先级任务 当多个任务同时产生时,按照任务设定等级优先被消费,例如a,b两种类型的job,优秀消费a,然后再消费b3. 安装3.1 源码运行配置文件位于gmq/conf.ini,可以根据自己项目需求修改配置 cd $GOPATH/src # 进入gopath/src目录git clone https://github.com/wuzhc/gmq.gitcd gmqgo get -u -v github.com/kardianos/govendor # 如果有就不需要安装了govendor sync -v # 如果很慢,可能需要翻墙go run main.go start3.2 执行文件运行cd $GOPATH/src/gmq# 编译成可执行文件go build # 启动./gmq start# 停止./gmq stop# 守护进程模式启动,不输出日志到consolenohup ./gmq start >/dev/null 2>&1 &# 守护进程模式下查看日志输出(配置文件conf.ini需要设置target_type=file,filename=gmq.log)tail -f gmq.log4. 客户端目前只实现python,go,php语言的客户端的demo,参考:https://github.com/wuzhc/demo/tree/master/mq 运行# php# 生产者php producer.php# 消费者php consumer.php# python# 生产者python producer.py# 消费者python consumer.py一条消息结构{ "id": "xxxx", # 任务id,这个必须是一个唯一值,将作为redis的缓存键 "topic": "xxx", # topic是一组job的分类名,消费者将订阅topic来消费该分类下的job "body": "xxx", # 消息内容 "delay": "111", # 延迟时间,单位秒 "TTR": "11111", # 执行超时时间,单位秒 "status": 1, # job执行状态,该字段由gmq生成 "consumeNum":1, # 被消费的次数,主要记录TTR>0时,被重复消费的次数,该字段由gmq生成}延迟任务 $data = [ 'id' => 'xxxx_id' . microtime(true) . rand(1,999999999), 'topic' => ["topic_xxx"], 'body' => 'this is a rpc test', 'delay' => '1800', // 单位秒,半个小时后执行 'TTR' => '0' ];超时任务 $data = [ 'id' => 'xxxx_id' . microtime(true) . rand(1,999999999), 'topic' => ["topic_xxx"], 'body' => 'this is a rpc test', 'delay' => '0', 'TTR' => '100' // 100秒后还未得到消费者ack确认,则再次添加到队列,将再次被被消费 ];异步任务$data = [ 'id' => 'xxxx_id' . microtime(true) . rand(1,999999999), 'topic' => ["topic_xxx"], 'body' => 'this is a rpc test', 'delay' => '0', 'TTR' => '0' ];优先级任务$data = [ 'id' => 'xxxx_id' . microtime(true) . rand(1,999999999), 'topic' => ["topic_A","topic_B","topic_C"], //优先消费topic_A,消费完后消费topic_B,最后再消费topic_C 'body' => 'this is a rpc test', 'delay' => '0', 'TTR' => '0' ];5. gmq流程图如下: ...

前言基础知识就像是一座大楼的地基，它决定了我们的技术高度。我们应该多掌握一些可移值的技术或者再过十几年应该都不会过时的技术，数据结构与算法就是其中之一。栈、队列、链表、堆 是数据结构与算法中的基础知识，是程序员的地基。 笔者写的 JavaScript 数据结构与算法之美 系列用的语言是 JavaScript ，旨在入门数据结构与算法和方便以后复习。 1. 线性表与非线性表线性表（Linear List）：就是数据排成像一条线一样的结构。每个线性表上的数据最多只有前和后两个方向。数组、链表、队列、栈 等就是线性表结构。 非线性表：数据之间并不是简单的前后关系。二叉树、堆、图 就是非线性表。 本文主要讲线性表，非线性表会在后面章节讲。 2. 数组 定义数组 (Array) 是一个有序的数据集合，我们可以通过数组名称 (name) 和索引 (index) 进行访问。数组的索引是从 0 开始的。特点数组是用一组连续的内存空间来存储的。所以数组支持 随机访问，根据下标随机访问的时间复杂度为 O(1)。 低效的插入和删除。数组为了保持内存数据的连续性，会导致插入、删除这两个操作比较低效，因为底层通常是要进行大量的数据搬移来保持数据的连续性。插入与删除的时间复杂度如下：插入：从最好 O(1) ，最坏 O(n) ，平均 O(n)删除：从最好 O(1) ，最坏 O(n) ，平均 O(n) 注意但是因为 JavaScript 是弱类型的语言，弱类型则允许隐式类型转换。 隐式：是指源码中没有明显的类型转换代码。也就是说，一个变量，可以赋值字符串，也可以赋值数值。 let str = "string"str = 123 console.log(str) // 123你还可以直接让字符串类型的变量和数值类型的变量相加，虽然得出的最终结果未必是你想象的那样，但一定不会报错。 let a = 123let b = "456"let c = a + b// 数值加字符串，结果是字符串console.log(c) // "123456"数组的每一项可以是不同的类型，比如： ...

队列栈与一般线性表区别线性表抽象是存储具有先后顺序元素数据的结构，支持任意位置的插入，删除操作。队列和栈限制插入删除操作，队列只能从尾部插入，首部取出(删除)，既先入先出;栈限制插入和取出操作只能在尾部进行，既先入后出。 实现方式队列和栈同一般线性表相同，可用数组和链式结构体实现。由于限制了插入和取出的位置，没有频繁的中间元素操作，数组实现比链式实现效率更高。对应缺点为初始化时要定义数组大小，无法动态分配大小。 代码实现栈struct stack;typedef struct stack sStack;typedef sStack *pStack;#define EmptyTOS -1;struct stack { int capacity; int topOfStack; long long int *data;};pStack elrInitStackInt(int capaticy) { pStack s; s = malloc(sizeof(sStack)); if (s == NULL) { printf("out of space!"); } s->data = malloc(sizeof(long long int) * capaticy); if (s->data == NULL) { printf("out of space!"); } s->capacity = capaticy; elrMakeStackEmpty(s); return s;}void elrFreeStackInt(pStack stack) { if (stack != NULL) { free(stack->data); free(stack); }}int elrIsStackEmpty(pStack stack) { return stack->topOfStack == EmptyTOS;}int elrIsStackFull(pStack stack) { return stack->topOfStack == (stack->capacity - 1);}void elrMakeStackEmpty(pStack stack) { stack->topOfStack = EmptyTOS;}void elrPushStackInt(pStack stack, long long int data) { if (elrIsStackFull(stack)) { printf("full stack"); } else { stack->data[++stack->topOfStack] = data; }}long long int elrPopStackInt(pStack stack) { if (elrIsStackEmpty(stack)) { printf("empty stack"); } else { return stack->data[--stack->topOfStack]; }}队列struct queue;typedef struct queue sQueue;typedef sQueue *pQueue;struct queue { int capacity; int front; int rear; int size; long long int *data;};pQueue elrInitQueueInt(int capaticy) { pQueue s; s = malloc(sizeof(sQueue)); if (s == NULL) { printf("out of space!"); } s->data = malloc(sizeof(long long int) * capaticy); if (s->data == NULL) { printf("out of space!"); } s->capacity = capaticy; elrMakeQueueEmpty(s); return s;}void elrFreeQueueInt(pQueue queue) { if (queue != NULL) { free(queue->data); free(queue); }}int elrIsQueueEmpty(pQueue queue) { return queue->size == 0;}int elrIsQueueFull(pQueue queue) { return queue->size == queue->capacity;}void elrMakeQueueEmpty(pQueue queue) { queue->size = 0; queue->front = 1; queue->rear = 0;}int succ(pQueue queue, int value) { if (++value == queue->capacity) { value = 0; } return value;}void elrEnqueuekInt(pQueue queue, long long int data) { if (elrIsQueueFull(queue)) { printf("full queue"); } else { queue->size++; queue->rear = succ(queue, queue->rear); queue->data[queue->rear] = data; }}long long int elrDequeueInt(pQueue queue) { if (elrIsQueueEmpty(queue)) { printf("empty queue"); } else { queue->size--; int data = queue->data[queue->front]; queue->front = succ(queue, queue->front); }}

数据结构-队列定义队列（queue）在计算机科学中，是一种先进先出的线性表。它只允许在表的前端进行删除操作，而在表的后端进行插入操作。进行插入操作的端称为队尾，进行删除操作的端称为队头。队列中没有元素时，称为空队列。 基于自定义数组实现的队列新建queue接口，用来规范所有queue子类package com.datastructure.queue;import java.awt.*;/** * @program: test * @description: * @author: Mr.Yang * @create: 2019-05-03 16:44 **/public class LoopQueue<E> implements Queue<E> { private E[] data; //指向队列的第一个元素，初始指向0 private int front; //指向队列的最后一个元素的后一个位置，初始指向0 private int tail; private int size; public LoopQueue(int capacity){ data = (E[]) new Object[capacity+1]; front=0; tail=0; size=0; } public LoopQueue(){ this(10); } /** * 因为容量放的时候多了个1，所以get容量的时候，需要减1 * @return */ public int getCapacity(){ return data.length-1; } /** * 1.if((tail + 1) % data.length == front) 如果tail + 1 超过了data.length的大小， * 代表当前tail指向已经超出了容量的大小，因为是循环式，所以需要tail去循环头元素中查看值是否有被占用， * 如果 == front 代表循环头没有，就需要扩容了。 * 2.举例： 元素容量为8，tail目前指向7 front 指向2 * if((7 + 1) % 8 == 2 ) if(0 == 2) 这里是false，因为front指向了2，所以代表 第0,1位是没有值的 * 所以这个值需要在在第0位放(空间利用) * 3.data[tail] = param tail当前指向的地方需要赋值，然后tail自增 循环体 的1，size+1 * @param param */ @Override public void enqueue(E param) { if((tail + 1) % data.length == front){ resize(getCapacity() * 2); } data[tail] = param ; tail = (tail + 1) % data.length; size ++ ; } /** * 扩充队列的容量 * 1.front代表了当前元素初始位置的指向 * 2.newData的第i位元素，应该等于 i + front % data.length 的值 * 3.举例：元素容量20，i 等于 0 ，front 等于 2，结果： newData[0] = data[(0 + 2) % 20] * = data[2] 意思就是，newData的第一位元素，应该等于data有值的第一位元素 * % data.length 的原因主要是为了防止数组越界错误 * 4.新数组赋值完成需要将 front 重新指向0，因为新数组的front指针是从0开始的。 * tail最后要指向等于size大小的值， * @param newCapacity */ private void resize(int newCapacity) { E[] newData = (E[]) new Object[newCapacity + 1]; for(int i = 0 ; i < size ; i++){ newData[i] = data[(i + front ) % data.length]; } data=newData; front = 0 ; tail = size; } /** * 1.如果队列为空抛出异常 * 2.用ret变量来接受当前队列头的值 * 3.接收成功之后将，队列头元素置空 * 4.front指针指向下一个元素 * 5.size大小-1 * 6.如果size大小占据了容量的1/4和size为容量的1/2且不等于0的时候，对容量进行缩减，缩减为原来容量的1/2 * 7.返回ret变量 * @return */ @Override public E dequeue() { if(isEmpty()){ throw new IllegalArgumentException("dequeue is fail ,because queue is empty"); } E ret = data[front]; data[front] = null; front = (front + 1) % data.length; size -- ; if(size == getCapacity() / 4 && getCapacity() / 2 != 0 ){ resize(getCapacity() / 2 ); } return ret; } @Override public E getFront() { if(isEmpty()){ throw new IllegalArgumentException("queue is empty"); } return data[front]; } @Override public int getSize() { return size; } /** * 当front和tail的值相等时，队列为空，初始两个指向的是同一个值（只有初始的时候，指向的是同一个地方） * @return */ @Override public boolean isEmpty() { return front == tail; } /** * 1.元素从 front位置开始循环遍历，i的值不能等于tail， * 也就是到tail的前一位，i = i + 1 且%data.length， * 因为i有可能从循环头重新开始 * 2.( i + 1 ) % data.length != tail 如果当前i + 1 % data.length * 不等于tail表示不到最后一个元素，就拼接， * @return */ @Override public String toString(){ StringBuffer sb = new StringBuffer(); sb.append(String.format("Array: size = %d , capacity = %d\n", size, getCapacity())); sb.append("front ["); for (int i = front; i != tail ; i = (i + 1) % data.length) { sb.append(data[i]); if (( i + 1 ) % data.length != tail) { sb.append(", "); } } sb.append("] tail"); return sb.toString(); }}新建ArrayQueue实现类package com.datastructure.queue;import com.datastructure.array.Array;/** * @program: test * @description: * @author: Mr.Yang * @create: 2019-05-03 18:19 **/public class ArrayQueue<E> implements Queue<E>{ Array<E> array; public ArrayQueue(int capacity){ array=new Array<E>(capacity); } public ArrayQueue(){ array=new Array<E>(); } @Override public void enqueue(E param) { array.addLast(param); } @Override public E dequeue() { return array.removeFirst(); } @Override public E getFront() { return array.getFirst(); } @Override public int getSize() { return array.getSize(); } @Override public boolean isEmpty() { return array.isEmpty(); } @Override public String toString(){ StringBuffer sb = new StringBuffer(); sb.append("front: "); sb.append("["); for(int i=0;i<array.getSize();i++){ sb.append(array.get(i)); if(i!=array.getSize()-1){ sb.append(", "); } } sb.append("] tail"); return sb.toString(); }}测试类package com.datastructure.queue;/** * @program: test * @description: * @author: Mr.Yang * @create: 2019-05-03 18:26 **/public class QueueTest { public static void main(String[] args) { ArrayQueue<Integer> integerArrayQueue = new ArrayQueue<>(); for (int i = 0; i < 10; i++) { integerArrayQueue.enqueue(i); System.out.println(integerArrayQueue); if(i % 3 == 2){ integerArrayQueue.dequeue(); System.out.println(integerArrayQueue); } } }}测试结果front: [0] tailfront: [0, 1] tailfront: [0, 1, 2] tailfront: [1, 2] tailfront: [1, 2, 3] tailfront: [1, 2, 3, 4] tailfront: [1, 2, 3, 4, 5] tailfront: [2, 3, 4, 5] tailfront: [2, 3, 4, 5, 6] tailfront: [2, 3, 4, 5, 6, 7] tailfront: [2, 3, 4, 5, 6, 7, 8] tailfront: [3, 4, 5, 6, 7, 8] tailfront: [3, 4, 5, 6, 7, 8, 9] tail可以看到测试结果是正确的，也符合队列结构的数据存取，但是因为是基于自定义数组来实现的，所以会调用数组方法的removeFirst方法，删除第一个元素的同时，会重新将后面所有元素前移，索引前移，均摊时间复杂度为O(n)。 ...

引言本周在编写短信验证码频率限制切面的时候，经潘老师给的实现思路，使用队列进行实现。 看了看java.util包下的Queue接口，发现还从来没用过呢！ Collection集合类接口，由它派生出List、Set和Queue，Map属于另一个独立的接口，和Collection没有继承关系。 List、Set和Map我们用的都是已经相当熟练了，今天，我们就来盘这个队列Queue！ 探索队列与栈都是数据结构的基础话题，队列：先进先出；栈：后进先出。 方法Queue接口中声明了六个方法，分成三对来使用。 入队操作 方法特点建议add入队失败抛出异常 offer入队失败返回false推荐出队操作 方法特点建议remove出队失败抛出异常 poll出队失败返回null推荐取队头操作 方法特点建议element队列为空时抛出异常 peek队列为空时返回null推荐PriorityQueue在java.util包中，除抽象类外，直接实现Queue接口的只有PriorityQueue优先级队列。 优先级队列比普通的队列要高级，普通的队列如果是先进的肯定是在队头的，而优先级队列根据优先级判断当前队头元素是什么。很适合实现操作系统中的按优先级实现进程调度。 如果需要使用优先级队列进行排序时，需要传入比较器。 该队列使用数组实现，线程不安全。 Deque java.util包中，Deque接口继承Queue接口。 Deque：double-ended queue，双端队列。 双端队列，相比普通队列就是可操作两端，有两个队头，也有两个队尾。 所以再去看Deque接口中声明的方法，都是两套的。offerFirst、offerLast、pollFirst、pollLast等。 所以说，如果使用双端队列，不仅可以当队列用，也可以当栈用，因为可以自己控制出的是队头还是队尾。 Deque有两个实现类：ArrayDeque和LinkedList。 原来LinkedList不仅实现了List接口，还实现了Deque接口。 两者的区别显而易见，一个是数组方式实现的，一个是链表的方式实现的。 TODO这些都是java.util包下的，都是线程不安全的实现，JDK所有线程安全的队列实现都在java.util.concurrent包下，也就是阻塞队列，以后再完善。

作者：个推平台研发工程师 祥子一、业务背景在个推的推送场景中，消息队列在整个系统中占有非常重要的位置。当 APP 有推送需求的时候, 会向个推发送一条推送命令，接到推送需求后，我们会把APP要求推送消息的用户放入下发队列中，进行消息下发；当同时有多个APP进行消息下发时，难免会出现资源竞争的情况, 因此就产生了优先级队列的需求，在下发资源固定的情况下, 高优先级的用户需要有更多的下发资源。二、基于 Kafka 的优先级队列方案针对以上场景，个推基于 Kafka 设计了第一版的优先级队列方案。Kafka 是 LinkedIn 开发的一个高性能、分布式消息系统；Kafka 在个推有非常广泛的应用，如日志收集、在线和离线消息分发等。架构在该方案中，个推将优先级统一设定为高、中、低三个级别。具体操作方案如下：对某个优先级根据 task (单次推送任务)维度，存入不同的 Topic，一个 task 只写入一个 Topic，一个 Topic 可存多个 task；消费模块根据优先级配额(如 6:3:1)，获取不同优先级的消息数，同一优先级轮询获取消息；这样既保证了高优先级用户可以更快地发送消息，又避免了低优先级用户出现没有下发的情况。Kafka 方案遇到的问题随着个推业务的不断发展，接入的 APP 数量逐渐增多，第一版的优先级方案也逐渐暴露出一些问题：当相同优先级的 APP 在同一时刻推送任务越来越多时，后面进入的 task 消息会因为前面 task 消息还存在队列情况而出现延迟。如下图所示, 当 task1 消息量过大时，在task1 消费结束前，taskN 将一直处于等待状态。Kafka 在 Topic 数量由 64 增长到 256 时，吞吐量下降严重，Kafka 的每个 Topic、每个分区都会对应一个物理文件。当 Topic 数量增加时，消息分散的落盘策略会导致磁盘 IO 竞争激烈，因此我们不能仅通过增加 Topic 数量来缓解第一点中的问题。基于上述问题，个推进行了新一轮的技术选型, 我们需要可以创建大量的 Topic, 同时吞吐性能不能比 Kafka 逊色。经过一段时间的调研，Apache Pulsar 引起了我们的关注。三、为什么是 PulsarApache Pulsar 是一个企业级的分布式消息系统，最初由 Yahoo 开发，在 2016 年开源，并于2018年9月毕业成为 Apache 基金会的顶级项目。Pulsar 已经在 Yahoo 的生产环境使用了三年多，主要服务于Mail、Finance、Sports、 Flickr、 the Gemini Ads platform、 Sherpa (Yahoo 的 KV 存储)。架构Topic 数量Pulsar 可以支持百万级别 Topic 数量的扩展，同时还能一直保持良好的性能。Topic 的伸缩性取决于它的内部组织和存储方式。Pulsar 的数据保存在 bookie (BookKeeper 服务器)上，处于写状态的不同 Topic 的消息，在内存中排序，最终聚合保存到大文件中，在 Bookie 中需要更少的文件句柄。另一方面 Bookie 的 IO 更少依赖于文件系统的 Pagecache，Pulsar 也因此能够支持大量的主题。消费模型Pulsar 支持三种消费模型：Exclusive、Shared 和Failover。Exclusive (独享)：一个 Topic 只能被一个消费者消费。Pulsar 默认使用这种模式。Shared(共享)：共享模式，多个消费者可以连接到同一个 Topic，消息依次分发给消费者。当一个消费者宕机或者主动断开连接时，那么分发给这个消费者的未确认(ack)的消息会得到重新调度，分发给其他消费者。Failover (灾备)：一个订阅同时只有一个消费者，可以有多个备份消费者。一旦主消费者故障，则备份消费者接管。不会出现同时有两个活跃的消费者。Exclusive和Failover订阅，仅允许一个消费者来使用和消费每个订阅的Topic。这两种模式都按 Topic 分区顺序使用消息。它们最适用于需要严格消息顺序的流(Stream)用例。Shared 允许每个主题分区有多个消费者。同一个订阅中的每个消费者仅接收Topic分区的一部分消息。Shared最适用于不需要保证消息顺序队列(Queue)的使用模式，并且可以按照需要任意扩展消费者的数量。存储Pulsar 引入了 Apache BookKeeper 作为存储层，BookKeeper 是一个专门为实时系统优化过的分布式存储系统，具有可扩展、高可用、低延迟等特性。具体介绍，请参考 BookKeeper官网。SegmentBookKeeper以 Segment (在 BookKeeper 内部被称作 ledger) 作为存储的基本单元。从 Segment 到消息粒度，都会均匀分散到 BookKeeper 的集群中。这种机制保证了数据和服务均匀分散在 BookKeeper 集群中。Pulsar 和 Kafka 都是基于 partition 的逻辑概念来做 Topic 存储的。最根本的不同是，Kafka 的物理存储是以 partition 为单位的，每个 partition 必须作为一个整体(一个目录)存储在某个 broker 上。 而 Pulsar 的 partition 是以 segment 作为物理存储的单位，每个 partition 会再被打散并均匀分散到多个 bookie 节点中。这样的直接影响是，Kafka 的 partition 的大小，受制于单台 broker 的存储；而 Pulsar 的 partition 则可以利用整个集群的存储容量。扩容当 partition 的容量达到上限后，需要扩容的时候，如果现有的单台机器不能满足，Kafka 可能需要添加新的存储节点，并将 partition 的数据在节点之间搬移达到 rebalance 的状态。而 Pulsar 只需添加新的 Bookie 存储节点即可。新加入的节点由于剩余空间大，会被优先使用，接收更多的新数据；整个扩容过程不涉及任何已有数据的拷贝和搬移。Broker 故障Pulsar 在单个节点失败时也会体现同样的优势。如果 Pulsar 的某个服务节点 broker 失效，由于 broker 是无状态的，其他的 broker 可以很快接管 Topic，不会涉及 Topic 数据的拷贝；如果存储节点 Bookie 失效，在集群后台中，其他的 Bookie 会从多个 Bookie 节点中并发读取数据，并对失效节点的数据自动进行恢复，对前端服务不会造成影响。Bookie 故障Apache BookKeeper 中的副本修复是 Segment (甚至是 Entry)级别的多对多快速修复。这种方式只会复制必须的数据，这比重新复制整个主题分区要精细。如下图所示，当错误发生时， Apache BookKeeper 可以从 bookie 3 和 bookie 4 中读取 Segment 4 中的消息，并在 bookie 1 处修复 Segment 4。所有的副本修复都在后台进行，对 Broker 和应用透明。当某个 Bookie 节点出错时，BookKeeper会自动添加可用的新 Bookie 来替换失败的 Bookie，出错的 Bookie 中的数据在后台恢复，所有 Broker 的写入不会被打断，而且不会牺牲主题分区的可用性。四、基于 Pulsar 的优先级队列方案在设计思路上，Pulsar 方案和 Kafka 方案并没有多大区别。但在新方案中，个推技术团队借助 Pulsar 的特性，解决了 Kafka 方案中存在的问题。根据 task 动态生成 Topic，保证了后进入的 task 不会因为其他 task 消息堆积而造成等待情况。中高优先级 task 都独享一个 Topic，低优先级 task 共享 n 个 Topic。相同优先级内，各个 task 轮询读取消息，配额满后流转至下一个优先级。相同优先级内, 各个 task 可动态调整 quota， 在相同机会内，可读取更多消息。利用 Shared 模式, 可以动态添加删除 consumer，且不会触发 Rebalance 情况。利用 BookKeeper 特性，可以更灵活的添加存储资源。五、Pulsar 其他实践不同 subscription 之间相对独立，如果想要重复消费某个 Topic 的消息，需要使用不同的 subscriptionName 订阅；但是一直增加新的 subscriptionName，backlog 会不断累积。如果 Topic 无人订阅，发给它的消息默认会被删除。因此如果 producer 先发送，consumer 后接收，一定要确保 producer 发送之前，Topic 有 subscription 存在(哪怕 subscribe 之后 close 掉)，否则这段时间发送的消息会导致无人处理。如果既没有人发送消息，又没有人订阅消息，一段时间后 Topic 会自动删除。Pulsar 的 TTL 等设置，是针对整个 namespace 起效的，无法针对单个 Topic。Pulsar 的键都建立在 zookeeper 的根目录上，在初始化时建议增加总节点名。目前 Pulsar 的 java api 设计，消息默认需要显式确认，这一点跟 Kafka 不一样。Pulsar dashboard 上的 storage size 和 prometheus 上的 storage size (包含副本大小)概念不一样。把dbStorage_rocksDB_blockCacheSize 设置的足够大；当消息体量大，出现backlog 大量堆积时, 使用默认大小(256M)会出现读耗时过大情况，导致消费变慢。使用多 partition，提高吞吐。在系统出现异常时，主动抓取 stats 和 stats-internal，里面有很多有用数据。如果业务中会出现单 Topic 体量过大的情况，建议把 backlogQuotaDefaultLimitGB 设置的足够大(默认10G), 避免因为默认使用producer_request_hold 模式出现 block producer 的情况；当然可以根据实际业务选择合适的 backlogQuotaDefaultRetentionPolicy。根据实际业务场景主动选择 backlog quota。prometheus 内如果发现读耗时为空情况，可能是因为直接读取了缓存数据；Pulsar 在读取消息时会先读取 write cache, 然后读取 read cache；如果都没有命中, 则会在 RocksDB 中读取条目位子后，再从日志文件中读取该条目。写入消息时, Pulsar 会同步写入 journal 和 write cache；write cache 再异步写入日志文件和 RocksDB； 所以有资源的话，建议 journal 盘使用SSD。六、总结现在, 个推针对优先级中间件的改造方案已经在部分现网业务中试运行，对于 Pulsar 的稳定性，我们还在持续关注中。作为一个2016 年才开源的项目，Pulsar 拥有非常多吸引人的特性，也弥补了其他竞品的短板，例如跨地域复制、多租户、扩展性、读写隔离等。尽管在业内使用尚不广泛, 但从现有的特性来说, Pulsar 表现出了取代 Kafka 的趋势。在使用 Pulsar 过程中，我们也遇到了一些问题, 在此特别感谢翟佳和郭斯杰(两位均为 Stream Native 的核心工程师、开源项目 Apache Pulsar 的 PMC 成员)给我们提供的支持和帮助。参考文献：[1] 比拼 Kafka, 大数据分析新秀Pulsar 到底好在哪(https://www.infoq.cn/article/...[2] 开源实时数据处理系统Pulsar：一套搞定Kafka+Flink+DB(https://juejin.im/post/5af414… ...

前言较长一段时间以来我都发现不少开发者对 jdk 中的 J.U.C（java.util.concurrent）也就是 Java 并发包的使用甚少，更别谈对它的理解了；但这却也是我们进阶的必备关卡。之前或多或少也分享过相关内容，但都不成体系；于是便想整理一套与并发包相关的系列文章。其中的内容主要包含以下几个部分：根据定义自己实现一个并发工具。JDK 的标准实现。实践案例。基于这三点我相信大家对这部分内容不至于一问三不知。既然开了一个新坑，就不想做的太差；所以我打算将这个列表下的大部分类都讲到。所以本次重点讨论 ArrayBlockingQueue。自己实现在自己实现之前先搞清楚阻塞队列的几个特点：基本队列特性：先进先出。写入队列空间不可用时会阻塞。获取队列数据时当队列为空时将阻塞。实现队列的方式多种，总的来说就是数组和链表；其实我们只需要搞清楚其中一个即可，不同的特性主要表现为数组和链表的区别。这里的 ArrayBlockingQueue 看名字很明显是由数组实现。我们先根据它这三个特性尝试自己实现试试。初始化队列我这里自定义了一个类：ArrayQueue，它的构造函数如下： public ArrayQueue(int size) { items = new Object[size]; }很明显这里的 items 就是存放数据的数组；在初始化时需要根据大小创建数组。写入队列写入队列比较简单，只需要依次把数据存放到这个数组中即可，如下图：但还是有几个需要注意的点：队列满的时候，写入的线程需要被阻塞。写入过队列的数量大于队列大小时需要从第一个下标开始写。先看第一个队列满的时候，写入的线程需要被阻塞，先来考虑下如何才能使一个线程被阻塞，看起来的表象线程卡住啥事也做不了。有几种方案可以实现这个效果:Thread.sleep(timeout)线程休眠。object.wait() 让线程进入 waiting 状态。当然还有一些 join、LockSupport.part 等不在本次的讨论范围。阻塞队列还有一个非常重要的特性是：当队列空间可用时（取出队列），写入线程需要被唤醒让数据可以写入进去。所以很明显Thread.sleep(timeout)不合适，它在到达超时时间之后便会继续运行；达不到空间可用时才唤醒继续运行这个特点。其实这样的一个特点很容易让我们想到 Java 的等待通知机制来实现线程间通信；更多线程见通信的方案可以参考这里：深入理解线程通信所以我这里的做法是，一旦队列满时就将写入线程调用 object.wait() 进入 waiting 状态，直到空间可用时再进行唤醒。 /** * 队列满时的阻塞锁 / private Object full = new Object(); /* * 队列空时的阻塞锁 */ private Object empty = new Object();所以这里声明了两个对象用于队列满、空情况下的互相通知作用。在写入数据成功后需要使用 empty.notify()，这样的目的是当获取队列为空时，一旦写入数据成功就可以把消费队列的线程唤醒。这里的 wait 和 notify 操作都需要对各自的对象使用 synchronized 方法块，这是因为 wait 和 notify 都需要获取到各自的锁。消费队列上文也提到了：当队列为空时，获取队列的线程需要被阻塞，直到队列中有数据时才被唤醒。代码和写入的非常类似，也很好理解；只是这里的等待、唤醒恰好是相反的，通过下面这张图可以很好理解：总的来说就是：写入队列满时会阻塞直到获取线程消费了队列数据后唤醒写入线程。消费队列空时会阻塞直到写入线程写入了队列数据后唤醒消费线程。测试先来一个基本的测试：单线程的写入和消费。3123123412345通过结果来看没什么问题。当写入的数据超过队列的大小时，就只能消费之后才能接着写入。2019-04-09 16:24:41.040 [Thread-0] INFO c.c.concurrent.ArrayQueueTest - [Thread-0]1232019-04-09 16:24:41.040 [main] INFO c.c.concurrent.ArrayQueueTest - size=32019-04-09 16:24:41.047 [main] INFO c.c.concurrent.ArrayQueueTest - 12342019-04-09 16:24:41.048 [main] INFO c.c.concurrent.ArrayQueueTest - 123452019-04-09 16:24:41.048 [main] INFO c.c.concurrent.ArrayQueueTest - 123456从运行结果也能看出只有当消费数据后才能接着往队列里写入数据。而当没有消费时，再往队列里写数据则会导致写入线程被阻塞。并发测试三个线程并发写入300条数据，其中一个线程消费一条。=====0299最终的队列大小为 299，可见线程也是安全的。由于不管是写入还是获取方法里的操作都需要获取锁才能操作，所以整个队列是线程安全的。ArrayBlockingQueue下面来看看 JDK 标准的 ArrayBlockingQueue 的实现，有了上面的基础会更好理解。初始化队列看似要复杂些，但其实逐步拆分后也很好理解：第一步其实和我们自己写的一样，初始化一个队列大小的数组。第二步初始化了一个重入锁，这里其实就和我们之前使用的 synchronized 作用一致的；只是这里在初始化重入锁的时候默认是非公平锁，当然也可以指定为 true 使用公平锁；这样就会按照队列的顺序进行写入和消费。更多关于 ReentrantLock 的使用和原理请参考这里：ReentrantLock 实现原理三四两步则是创建了 notEmpty notFull 这两个条件，他的作用于用法和之前使用的 object.wait/notify 类似。这就是整个初始化的内容，其实和我们自己实现的非常类似。写入队列其实会发现阻塞写入的原理都是差不多的，只是这里使用的是 Lock 来显式获取和释放锁。同时其中的 notFull.await();notEmpty.signal(); 和我们之前使用的 object.wait/notify 的用法和作用也是一样的。当然它还是实现了超时阻塞的 API。也是比较简单，使用了一个具有超时时间的等待方法。消费队列再看消费队列：也是差不多的，一看就懂。而其中的超时 API 也是使用了 notEmpty.awaitNanos(nanos) 来实现超时返回的，就不具体说了。实际案例说了这么多，来看一个队列的实际案例吧。背景是这样的：有一个定时任务会按照一定的间隔时间从数据库中读取一批数据，需要对这些数据做校验同时调用一个远程接口。简单的做法就是由这个定时任务的线程去完成读取数据、消息校验、调用接口等整个全流程；但这样会有一个问题：假设调用外部接口出现了异常、网络不稳导致耗时增加就会造成整个任务的效率降低，因为他都是串行会互相影响。所以我们改进了方案：其实就是一个典型的生产者消费者模型：生产线程从数据库中读取消息丢到队列里。消费线程从队列里获取数据做业务逻辑。这样两个线程就可以通过这个队列来进行解耦，互相不影响，同时这个队列也能起到缓冲的作用。但在使用过程中也有一些小细节值得注意。因为这个外部接口是支持批量执行的，所以在消费线程取出数据后会在内存中做一个累加，一旦达到阈值或者是累计了一个时间段便将这批累计的数据处理掉。但由于开发者的大意，在消费的时候使用的是 queue.take() 这个阻塞的 API；正常运行没啥问题。可一旦原始的数据源，也就是 DB 中没数据了，导致队列里的数据也被消费完后这个消费线程便会被阻塞。这样上一轮积累在内存中的数据便一直没机会使用，直到数据源又有数据了，一旦中间间隔较长时便可能会导致严重的业务异常。所以我们最好是使用 queue.poll(timeout) 这样带超时时间的 api，除非业务上有明确的要求需要阻塞。这个习惯同样适用于其他场景，比如调用 http、rpc 接口等都需要设置合理的超时时间。总结关于 ArrayBlockingQueue 的相关分享便到此结束，接着会继续更新其他并发容器及并发工具。对本文有任何相关问题都可以留言讨论。本文涉及到的所有源码：https://github.com/crossoverJ…你的点赞与分享是对我最大的支持 ...

你好，是我琉忆，PHP程序员面试笔试系列图书的作者。本周（2019.3.18至3.22）的一三五更新的文章如下：周一：PHP面试常考之数据结构——链表的概念周三：PHP面试常考之数据结构——栈和队列周五：PHP面试常考之数据结构——自己整理了一篇“PHP如何实现链表？”的文章，关注公众号：“琉忆编程库”，回复：“链表”，我发给你。一、链表链表（Linked list）是一种常见的基础数据结构，是一种线性表，但是并不会按线性的顺序存储数据，而是在每一个节点里存到下一个节点的指针(Pointer)。使用链表结构可以克服数组链表需要预先知道数据大小的缺点，链表结构可以充分利用计算机内存空间，实现灵活的内存动态管理。但是链表失去了数组随机读取的优点，同时链表由于增加了结点的指针域，空间开销比较大。链表有很三种不同的类型：单向链表，双向链表以及循环链表。二、单向链表单向链表包含两个域，一个信息域和一个指针域。这个链接指向列表中的下一个节点，而最后一个节点则指向一个空值。如图：三、双向链表每个节点有两个连接：一个指向前一个节点，（当此“连接”为第一个“连接”时，指向空值或者空列表）；而另一个指向下一个节点，（当此“连接”为最后一个“连接”时，指向空值或者空列表）如图：四、循环链表在一个循环链表中,首节点和末节点被连接在一起。这种方式在单向和双向链表中皆可实现。要转换一个循环链表，你开始于任意一个节点然后沿着列表的任一方向直到返回开始的节点。再来看另一种方法，循环链表可以被视为“无头无尾”。这种列表很利于节约数据存储缓存，假定你在一个列表中有一个对象并且希望所有其他对象迭代在一个非特殊的排列下。指向整个列表的指针可以被称作访问指针。自己编写的《PHP程序员面试笔试宝典》和《PHP程序员面试笔试真题解析》书籍，已在各大电商平台销售，两本可以帮助你更快更好的拿到offer的书。更多PHP相关的面试知识、考题可以关注公众号获取：琉忆编程库对本文有什么问题或建议都可以进行留言，我将不断完善追求极致，感谢你们的支持。

前言上一篇：算法分析下一篇：基本排序本篇内容主要是栈，队列 (和包)的基本数据类型和数据结构在很多应用中，我们需要维护多个对象的集合，而对这个集合的操作也很简单基本数据类型对象的集合操作：insert – 向集合中添加新的对象remove – 去掉集合中的某个元素iterate – 遍历集合中的元素并对他们执行某种操作test if empty – 检查集合是否为空做插入和删除操作时我们要明确以什么样的形式去添加元素，或我们要删除集合中的哪个元素。处理这类问题有两个经典的基础数据结构：栈(stack) 和队列(queue)两者的区别在于去除元素的方式：栈：去除最近加入的元素，遵循后进先出原则(LIFO: last in first out)。插入元素对应的术语是入栈 – push；去掉最近加入的元素叫出栈 – pop队列：去除最开始加入的元素，遵循先进先出原则(FIFO: first in first out)。关注最开始加入队列的元素，为了和栈的操作区分，队列加入元素的操作叫做入队 – enqueue；去除元素的操作叫出队 – dequeue此篇隐含的主题是模块式编程，也是平时开发需要遵守的原则模块化编程这一原则的思想是将接口与实现完全分离。比如我们精确定义了一些数据类型和数据结构(如栈，队列等)，我们想要的是把实现这些数据结构的细节完全与客户端分离。客户端可以选择数据结构不同的实现方式，但是客户端代码只能执行基本操作。实现的部分无法知道客户端需求的细节，它所要做的只是实现这些操作，这样，很多不同的客户端都可以使用同一个实现，这使得我们能够用模块式可复用的算法与数据结构库来构建更复杂的算法和数据结构，并在必要的时候更关注算法的效率。Separate client and implementation via API.API：描述数据类型特征的操作Client：使用API操作的客户端程序。Implementation：实现API操作的代码。下面具体看下这两种数据结构的实现栈栈 API假设我们有一个字符串集合，我们想要实现字符串集合的储存，定期取出并且返回最后加入的字符串，并检查集合是否为空。我们需要先写一个客户端然后再看它的实现。字符串数据类型的栈性能要求：所有操作都花费常数时间客户端：从标准输入读取逆序的字符串序列测试客户端import edu.princeton.cs.algs4.StdIn;import edu.princeton.cs.algs4.StdOut;public static void main(String[] args){ StackOfStrings stack = new StackOfStrings(); while (!StdIn.isEmpty()) { //从标准输入获取一些字符串 String s = StdIn.readString(); //如果字符串为"-",则客户端将栈顶的字符串出栈，并打印出栈的字符串 if (s.equals("-")) StdOut.print(stack.pop()); //否则将字符串入栈到栈顶 else stack.push(s); }}客户端输入输出：栈的实现：链表链表(linked-list)连接待添加…我们想保存一个有节点组成的，用来储存字符串的链表。节点包含指向链表中下一个元素的引用(first).维持指针 first 指向链表中的第一个节点Push：入栈，在链表头插入一个新的节点Pop：出栈，去掉链表头处第一个节点Java 实现public class LinkedStackOfStrings{ //栈中唯一的实例变量是链表中的第一个节点的引用 private Node first = null; //内部类，节点对象，构成链表中的元素，由一个字符串和指向另一个节点的引用组成 private class Node { private String item; private Node next; } public boolean isEmpty() { return first == null; } // public void push(String item) { //将指向链表头的指针先保存 Node oldfirst = first; //创建新节点：我们将要插入表头的节点 first = new Node(); first.item = item; //实例变量的next指针指向链表oldfirst元素，现在变成链表的第二个元素 first.next = oldfirst; } //出栈 public String pop() { //将链表中的第一个元素储存在标量 item 中 String item = first.item; //去掉第一个节点：将原先指向第一个元素的指针指向下一个元素，然后第一个节点就等着被垃圾回收处理 first = first.next; //返回链表中原先保存的元素 return item; }}图示：出栈：入栈：性能分析通过分析提供给客户算法和数据结构的性能信息，评估这个实现对以不同客户端程序的资源使用量Proposition 在最坏的情况下，每个操作只需要消耗常数时间(没有循环)。Proposition 具有n个元素的栈使用 ~40n 个字节内存(没有考虑字符串本身的内存，因为这些空间的开销在客户端上)栈的实现：数组栈用链表是实现花费常数的时间，但是栈还有更快的实现另一种实现栈的 natural way 是使用数组储存栈上的元素将栈中的N个元素保存在数组中，索引为 n，n 对应的数组位置即为栈顶的位置，即下一个元素加入的地方使用数组 s[] 在栈上存储n个元素。push()：在 s[n] 处添加新元素。pop()：从 s[n-1] 中删除元素。在改进前使用数组的一个缺点是必须声明数组的大小，所以栈有确定的容量。如果栈上的元素个数比栈的容量多，我们就必须处理这个问题(调整数组)Java 实现public class FixedCapacityStackOfStrings{ private String[] s; //n 为栈的大小，栈中下一个开放位置，也为下一个元素的索引 private int n = 0; //int capacity：看以下说明 public FixedCapacityStackOfStrings(int capacity) { s = new String[capacity]; } public boolean isEmpty() { return n == 0; } public void push(String item) { //将元素放在 n 索引的位置，然后 n+1 s[n++] = item; } public String pop() { //然后返回数组n-1的元素 return s[–n]; }}int capacity： 在构造函数中加入了容量的参数，破坏了API，需要客户端提供栈的容量。不过实际上我们不会这么做，因为大多数情况下，客户端也无法确定需要多大栈，而且客户端也可能需要同时维护很多栈，这些栈又不同时间到达最大容量，同时还有其他因素的影响。这里只是为了简化。在调整数组中会处理可变容量的问题，避免溢出对于两种实现的思考上述的实现中我们暂时没有处理的问题：Overflow and underflowUnderflow ：客户端从空栈中出栈我们没有抛出异常Overflow ：使用数组实现，当客户端入栈超过容量发生栈溢出的问题Null item：客户端是否能像数据结构中插入空元素Loitering 对象游离：即在栈的数组中，我们有一个对象的引用，可是我们已经不再使用这个引用了数组中当我们减小 n 时，在数组中仍然有我们已经出栈的对象的指针，尽管我们不再使用它，但是Java系统并不知道。所以为了避免这个问题，有效地利用内存，最好将去除元素对应的项设为 null，这样就不会剩下旧元素的引用指针，接下来就等着垃圾回收机制去回收这些内存。这个问题比较细节化，但是却很重要。public String pop(){ String item = s[–n]; s[n] = null; return item;} 调整数组队列泛型迭代器栈与队列的应用附录 ...

这道题有点像优先队列的思想，简而言之就是挑选最小的入队处理，如果有多个队列就进行多个队列的处理；借鉴的思想是采用记录每个队列中的任务完成时间，然后在读入任务的时候进行轮询，选择结束时间最小的那个队列，然后进行处理和等待时间的计算；代码如下：#include <iostream>#include<stdlib.h>#include<stdio.h>#include <vector>#include <algorithm>using namespace std;struct node { int come, time;} tempcustomer;bool cmp1(node a, node b) { return a.come < b.come;}int main() { int n, k; scanf("%d%d", &n, &k); vector<node> custom; for(int i = 0; i < n; i++) { int hh, mm, ss, time; scanf("%d:%d:%d %d", &hh, &mm, &ss, &time); int cometime = hh * 3600 + mm * 60 + ss; if(cometime > 61200) continue; tempcustomer = {cometime, time * 60}; custom.push_back(tempcustomer); } sort(custom.begin(), custom.end(), cmp1); vector<int> window(k, 28800); double result = 0.0; for(int i = 0; i < custom.size(); i++) { int tempindex = 0, minfinish = window[0]; for(int j = 1; j < k; j++) { if(minfinish > window[j]) { minfinish = window[j]; tempindex = j; } } if(window[tempindex] <= custom[i].come) { window[tempindex] = custom[i].come + custom[i].time; } else { result += (window[tempindex] - custom[i].come); window[tempindex] += custom[i].time; } } if(custom.size() == 0) printf(“0.0”); else printf("%.1f", result / 60.0 / custom.size()); system(“pause”); return 0;} ...

题目用两个栈来实现一个队列，完成队列的Push和Pop操作。 队列中的元素为int类型。基本思路栈1:用于入队列存储栈2:出队列时将栈1的数据依次出栈，并入栈到栈2中栈2出栈即栈1的底部数据即队列要出的数据。注意:栈2为空才能补充栈1的数据，否则会打乱当前的顺序。代码const stack1 = [];const stack2 = [];function push(node){ stack1.push(node);}function pop(){ if(stack2.length === 0){ while(stack1.length>0){ stack2.push(stack1.pop()); } } return stack2.pop() || null;}

题目There are N rooms and you start in room 0. Each room has a distinct number in 0, 1, 2, …, N-1, and each room may have some keys to access the next room. Formally, each room i has a list of keys rooms[i], and each key rooms[i][j] is an integer in [0, 1, …, N-1] where N = rooms.length. A key rooms[i][j] = v opens the room with number v.Initially, all the rooms start locked (except for room 0). You can walk back and forth between rooms freely.Return true if and only if you can enter every room.Example 1:Input: [[1],[2],[3],[]]Output: trueExplanation: We start in room 0, and pick up key 1.We then go to room 1, and pick up key 2.We then go to room 2, and pick up key 3.We then go to room 3. Since we were able to go to every room, we return true.Example 2:Input: [[1,3],[3,0,1],[2],[0]]Output: falseExplanation: We can’t enter the room with number 2.Note:1 <= rooms.length <= 10000 <= rooms[i].length <= 1000The number of keys in all rooms combined is at most 3000.题目地址讲解这是一道典型的图的遍历，这里我采用了广度优先遍历，广度优先遍历可以用队列这种数据结构来实现。同时还需要一个Set用于记录已经访问过的room。感觉这道题还是比较简单的java代码class Solution { public boolean canVisitAllRooms(List<List<Integer>> rooms) { Set<Integer> set = new HashSet(); Queue<Integer> queue = new LinkedList<>(); for(Integer x:rooms.get(0)){ queue.offer(x); } while(!queue.isEmpty()){ int room = queue.poll(); if(set.contains(room)){ continue; }else{ set.add(room); List<Integer> list = rooms.get(room); if(list!=null && list.size()>0){ for(Integer x: list){ queue.offer(x); } } } } for(int i=1;i<rooms.size();i++){ if(!set.contains(i)){ return false; } } return true; }} ...

题目Write a class RecentCounter to count recent requests.It has only one method: ping(int t), where t represents some time in milliseconds.Return the number of pings that have been made from 3000 milliseconds ago until now.Any ping with time in [t - 3000, t] will count, including the current ping.It is guaranteed that every call to ping uses a strictly larger value of t than before.Example 1:Input: inputs = [“RecentCounter”,“ping”,“ping”,“ping”,“ping”], inputs = [[],[1],[100],[3001],[3002]]Output: [null,1,2,3,3]Note:Each test case will have at most 10000 calls to ping.Each test case will call ping with strictly increasing values of t.Each call to ping will have 1 <= t <= 10^9.题目地址讲解这道题最重要的是看懂题，每个ping都附带一个时间点，求这个时间点往前3000毫秒内的ping数量。这里我使用了一个队列来实现。Java代码class RecentCounter { private Queue<Integer> queue; public RecentCounter() { queue = new LinkedList<>(); } public int ping(int t) { while(queue.size()>0 && queue.peek()<t-3000){ queue.poll(); } queue.offer(t); return queue.size(); }}/** * Your RecentCounter object will be instantiated and called as such: * RecentCounter obj = new RecentCounter(); * int param_1 = obj.ping(t); */ ...

队列是先进先出（FIFO）的数据结构，插入操作叫做入队，只能添加在队列的末尾；删除操作叫做出队，只能移除第一个元素。在JS中，用数组可以很简单的实现队列。function Queue () { this.queue = [];}// 增加Queue.prototype.enQueue = function(x) { this.queue.push(x); return true;}// 删除Queue.prototype.deQueue = function() { if(this.isEmpty()) { return false; } this.queue.shift(); return true; }// 获取队首元素Queue.prototype.front = function() { if(this.isEmpty()) { return false; } this.queue[0]; }// 是否为空Queue.prototype.isEmpty = function() { return !this.queue.length}以上就实现了队列的数据结构，那么队列这种数据结构有什么作用呢？在广度优先搜索（BFS）中，很适合队列。那什么是BFS。在树的遍历中，有两种遍历方式，其中一种就是从根节点一层一层的往下遍历，这就是广度优先；另一种是先由根节点选一条路径直接遍历到叶子节点，这就是深度优先搜索（DFS）。队列可以用在BFS中，下面我们来实现一个广度优先搜索的例子，返回目标节点深度。 let root = { key: 1, children: [ { key:2, }, { key:3, children:[ { key:4, } ] } ] } // 数据源function bfs(root, target) { //利用上面创建的Queue，当然也可以直接用数组实现 let queue = new Queue(); let step = 0; // 根节点到目标节点之间的深度 queue.enQueue(root); //将根节点加入 //遍历队列 while(!queue.isEmpty()) { step += 1; let len = queue.length; // 分层遍历队列，没有目标元素则删除该层元素，继续遍历下一层 for(let i =0; i<len; i++) { let cur = queue.front() // 获取队首元素 if(target === cur.key) return step; //如果是目标元素，返回 // 如果不是，将下一层节点加入到队列 if(cur.children && cur.children.length) { cur.children.map(item => { queue.enQueue(item) }) } queue.deQueue() //然后将遍历过的节点删除， } }}bfs(root,4)这样我们就完成了BFS的实现思路，大家可已参照该思路在具体的业务中灵活运用BFS。 ...