轮询功能java实现

一、业务背景

由于项目需要满足大量轮询业务需求，需升级原有的轮询功能。改造过程中发现延时队列(DelayQueue)可以满足业务需求，但是论证后发现延时队列的查询、更新和删除时间复杂度为O(N)，因此要对延时队列进行优化。

二、延时队列介绍

延时队列使用优先队列(Priority)+延时器(Delay)实现，优先队列可以保证根节点为排序元素的最大值或者最小值，因此轮询实体在重写延时器(Delay)的CompaerTo方法后，可以保证优先队列的根节点始终为最近到期的轮询实体。

三、优化

使用HashMap索引轮询实体，从而使查询和更新的时间复杂度降为O(1)，但是删除的时间复杂度并不能优化到O(logn)，因为优先队列删除的时间复杂度为O(N)。如果使用红黑树来实现优先队列的功能可以做到删除时间复杂度为O(logN)，但是红黑数依然使用时间作为排序规则，所以并不能在这个维度下来做延时队列的删除。因此最终还是选用优先队列来实现轮询功能

四、代码实现

1、延时队列

import com.hey.bean.dto.ModemPollDto;
import java.util.HashMap;
import java.util.Map;
import java.util.PriorityQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;public class ModemPollQueue {private final transient ReentrantLock lock = new ReentrantLock();private final Map<String, ModemPollDto> map = new HashMap<>();private final PriorityQueue<ModemPollDto> q = new PriorityQueue<>();private Thread leader;private final Condition available = this.lock.newCondition();public boolean put(String k, ModemPollDto e) {ReentrantLock lock = this.lock;lock.lock();try {this.map.put(k, e);this.q.add(e);if (this.q.peek() == e) {this.leader = null;this.available.signal();} return true;} finally {lock.unlock();} }public ModemPollDto take() throws InterruptedException {ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (true) {ModemPollDto first = this.q.peek();if (first == null) {this.available.await();continue;} long delay = first.getDelay(TimeUnit.NANOSECONDS);if (delay <= 0L) {this.q.poll();this.map.remove(first.getModemId());return first;} first = null;if (this.leader != null) {this.available.await();continue;} Thread thisThread = Thread.currentThread();this.leader = thisThread;try {this.available.awaitNanos(delay);} finally {if (this.leader == thisThread)this.leader = null; } } } finally {if (this.leader == null && this.q.peek() != null)this.available.signal(); lock.unlock();} }public ModemPollDto get(Object o) {ReentrantLock lock = this.lock;lock.lock();try {return this.map.get(o);} finally {lock.unlock();} }public int size() {ReentrantLock lock = this.lock;lock.lock();try {return this.map.size();} finally {lock.unlock();} }public boolean remove(String modemId) {ReentrantLock lock = this.lock;lock.lock();try {ModemPollDto remove = this.map.remove(modemId);if (remove == null)return false; return this.q.remove(remove);} finally {lock.unlock();} }public ModemPollDto[] toArray() {ReentrantLock lock = this.lock;lock.lock();try {return this.q.<ModemPollDto>toArray(new ModemPollDto[this.q.size()]);} finally {lock.unlock();} }
}

2、轮询实体

import java.util.Date;
import java.util.Objects;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class ModemPollDto implements Delayed {private Integer id;private String modemName;private String modemId;private String modemIp;private Integer modemType;private Integer beamNum;private Integer polingNotWithdraw;private Integer polingSlot;private Integer currentPollCnt;private Integer pollMultiple;private Long time;private Long passTime;private Integer pollCount;private Integer pollResponseCount;public boolean equals(Object o) {if (this == o)return true; if (o == null || getClass() != o.getClass())return false; ModemPollDto that = (ModemPollDto)o;return Objects.equals(this.modemId, that.modemId);}public long getDelay(TimeUnit unit) {return this.time.longValue() - System.currentTimeMillis();}public int compareTo(Delayed o) {ModemPollDto item = (ModemPollDto)o;long diff = this.time.longValue() - item.time.longValue();if (diff < 0L)return -1; return 1;}
}

3、轮询类

public class RemPollJob {@Asyncpublic void poll() {ModemPollQueue queue = PollTaskConstant.remotePoll;while (true) {try {ModemPollDto take = queue.take();if (take.getCurrentPollCnt().equals(Integer.valueOf(take.getPolingSlot().intValue() - 1))) {addLeaveTheNetCaveat(take, take.getCurrentPollCnt().intValue());} else if (take.getCurrentPollCnt().intValue() >= take.getPolingSlot().intValue()) {if (take.getPolingNotWithdraw().intValue() == 0) {leave(take);continue;} } take.setCurrentPollCnt(Integer.valueOf(take.getCurrentPollCnt().intValue() + 1));take.setPollCount(Integer.valueOf(take.getPollCount().intValue() + 1));take.setPassTime(Long.valueOf(System.currentTimeMillis() - take.getTime().longValue()));take.setTime(Long.valueOf(System.currentTimeMillis() + (take.getPollMultiple().intValue() * 1000)));queue.put(take.getModemId(), take);issuePoll(take);} catch (Exception e) {e.printStackTrace();} } }
}

轮询功能java实现相关推荐

java 轮询请求_使用RxJava来实现网络请求轮询功能
原标题:使用RxJava来实现网络请求轮询功能近日有媒体报道称,腾讯重金入股永辉超市旗下生鲜超市超级物种,目前交易已经完成.受此刺激,永辉超市股价迅速涨停,午后临时停牌.若此举成行,超级物种将更有底 ...
web服务器（LAMP）通过DNS轮询功能和nfs共享实现负载均衡
web服务器(LAMP)通过DNS轮询功能和nfs共享实现负载均衡,部署discuz论坛拓扑思路: 服务器1:mariadb+nfs:172.20.120.40 服务器2:apache+php-fp ...
在verto_communicator中添加视频floor轮询功能
只是实现了简单的视频轮询功能,还需要修改完善!!! 1.在src/partials/chat.html中添加 <div><p>{{num}}</p><butt ...
ajax轮询数据库 java,ajax 轮询请求后台服务器
// var i=0; //声明轮询次数变量 $(document).ready(function(){ c = window.setInterval("getResult()", ...
java 轮询请求接口_js调用轮询接口
##### 项目中遇到需要很多个需要轮询处理的接口,然后简单的封装了下,做个记录,以后用到类似的直接copy ##### // polling-utils.js /** * @descripting ...
Vue项目实战04 : Vue 轮询接口的实现
项目中我们经常需要实现轮询-每隔几秒请求一次接口刷新数据一般都会使用setInterval,但要注意单纯使用它会导致页面卡死,所以一定要清除定时器 setInterval不会清除定时器队列,每重复执 ...
Linux C语言在用户态实现一个低时延通知（eventfd）+轮询（无锁队列ring）机制的消息队列
目录 fastq.c fastq.h test-0.c test-1.c https://github.com/Rtoax/test/tree/master/ipc/github/fastq fast ...
DPDK PMD（ Poll Mode Driver）轮询模式驱动程序
DPDK PMD( Poll Mode Driver)轮询模式驱动程序目录 Mellanox PMDs 轮询模式驱动程序要求和假设设计原则逻辑核心,内存和NIC队列关系设备标识,所有权和配置 ...
轮循码/轮巡码/轮询码
轮循码一个二维码自带轮询功能,多码轮询,降低高并发,功能强大. 为何我们推荐使用轮巡码现在单个商户收款码太过于频繁,容易触碰风控.根据这种情况我们推出轮巡码策略可以一对一定制, 一个二维码内置上万 ...