
DelayQueue泛型参数得实现Delayed接口,Delayed继承了Comparable接口。
getDelay方法返回这个任务还剩多久时间可以执行,小于0的时候说明可以这个延迟任务到了执行的时间了。compareTo这个是对任务排序的,保证最先到延迟时间的任务排到队列的头。@Getter
public class SanYouTask implements Delayed {
private final String taskContent;
private final Long triggerTime;
public SanYouTask(String taskContent, Long delayTime) {
this.taskContent = taskContent;
this.triggerTime = System.currentTimeMillis() + delayTime * 1000;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(triggerTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(Delayed o) {
return this.triggerTime.compareTo(((SanYouTask) o).triggerTime);
}
}@Slf4j
public class DelayQueueDemo {
public static void main(String[] args) {
DelayQueue<SanYouTask> sanYouTaskDelayQueue = new DelayQueue<>();
new Thread(() -> {
while (true) {
try {
SanYouTask sanYouTask = sanYouTaskDelayQueue.take();
log.info("获取到延迟任务:{}", sanYouTask.getTaskContent());
} catch (Exception e) {
}
}
}).start();
log.info("提交延迟任务");
sanYouTaskDelayQueue.offer(new SanYouTask("三友的java日记5s", 5L));
sanYouTaskDelayQueue.offer(new SanYouTask("三友的java日记3s", 3L));
sanYouTaskDelayQueue.offer(new SanYouTask("三友的java日记8s", 8L));
}
}
成功实现了延迟任务。
offer方法在提交任务的时候,会通过根据compareTo的实现对任务进行排序,将最先需要被执行的任务放到队列头。take方法获取任务的时候,会拿到队列头部的元素,也就是队列中最早需要被执行的任务,通过getDelay返回值判断任务是否需要被立刻执行,如果需要的话,就返回任务,如果不需要就会等待这个任务到延迟时间的剩余时间,当时间到了就会将任务返回。@Slf4j
public class TimerDemo {
public static void main(String[] args) {
Timer timer = new Timer();
log.info("提交延迟任务");
timer.schedule(new TimerTask() {
@Override
public void run() {
log.info("执行延迟任务");
}
}, 5000);
}
}
public abstract class TimerTask implements Runnable {
//忽略其它属性
long nextExecutionTime;
}@Slf4j
public class ScheduledThreadPoolExecutorDemo {
public static void main(String[] args) {
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2, new ThreadPoolExecutor.CallerRunsPolicy());
log.info("提交延迟任务");
executor.schedule(() -> log.info("执行延迟任务"), 5, TimeUnit.SECONDS);
}
}
在提交延迟任务的时候,任务会被封装一个任务会被封装成ScheduledFutureTask对象,然后放到DelayedWorkQueue阻塞队列中。
ScheduledFutureTaskScheduledFutureTask实现了前面提到的Delayed接口,所以其实可以猜到DelayedWorkQueue会根据ScheduledFutureTask对于Delayed接口的实现来排序,所以线程能够获取到最早到延迟时间的任务。当线程从DelayedWorkQueue中获取到需要执行的任务之后就会执行任务。
当发送消息的时候只需要指定延迟等级即可。如果这18个等级的延迟时间不符和你的要求,可以修改RocketMQ服务端的配置文件。<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.1</version>
<!--web依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<version>2.2.5.RELEASE</version>
</dependency>rocketmq:
name-server: 192.168.200.144:9876 #服务器ip:nameServer端口
producer:
group: sanyouProducer@RestController
@Slf4j
public class RocketMQDelayTaskController {
@Resource
private DefaultMQProducer producer;
@GetMapping("/rocketmq/add")
public void addTask(@RequestParam("task") String task) throws Exception {
Message msg = new Message("sanyouDelayTaskTopic", "TagA", task.getBytes(RemotingHelper.DEFAULT_CHARSET));
msg.setDelayTimeLevel(2);
// 发送消息并得到消息的发送结果,然后打印
log.info("提交延迟任务");
producer.send(msg);
}
}@Component
@RocketMQMessageListener(consumerGroup = "sanyouConsumer", topic = "sanyouDelayTaskTopic")
@Slf4j
public class SanYouDelayTaskTopicListener implements RocketMQListener<String> {
@Override
public void onMessage(String msg) {
log.info("获取到延迟任务:{}", msg);
}
}
生产者发送延迟消息之后,RocketMQ服务端在接收到消息之后,会去根据延迟级别是否大于0来判断是否是延迟消息
定时任务RocketMQ这种实现方式相比于前面提到的三种更加可靠,因为前面提到的三种任务内容都是存在内存的,服务器重启任务就丢了,如果要实现任务不丢还得自己实现逻辑,但是RocketMQ消息有持久化机制,能够保证任务不丢失。<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>2.2.5.RELEASE</version>
</dependency>spring:
rabbitmq:
host: 192.168.200.144 #服务器ip
port: 5672
virtual-host: /@Configuration
public class RabbitMQConfiguration {
@Bean
public DirectExchange sanyouDirectExchangee() {
return new DirectExchange("sanyouDirectExchangee");
}
@Bean
public Queue sanyouQueue() {
return QueueBuilder
//指定队列名称,并持久化
.durable("sanyouQueue")
//设置队列的超时时间为5秒,也就是延迟任务的时间
.ttl(5000)
//指定死信交换机
.deadLetterExchange("sanyouDelayTaskExchangee")
.build();
}
@Bean
public Binding sanyouQueueBinding() {
return BindingBuilder.bind(sanyouQueue()).to(sanyouDirectExchangee()).with("");
}
@Bean
public DirectExchange sanyouDelayTaskExchange() {
return new DirectExchange("sanyouDelayTaskExchangee");
}
@Bean
public Queue sanyouDelayTaskQueue() {
return QueueBuilder
//指定队列名称,并持久化
.durable("sanyouDelayTaskQueue")
.build();
}
@Bean
public Binding sanyouDelayTaskQueueBinding() {
return BindingBuilder.bind(sanyouDelayTaskQueue()).to(sanyouDelayTaskExchange()).with("");
}
}@RestController
@Slf4j
public class RabbitMQDelayTaskController {
@Resource
private RabbitTemplate rabbitTemplate;
@GetMapping("/rabbitmq/add")
public void addTask(@RequestParam("task") String task) throws Exception {
// 消息ID,需要封装到CorrelationData中
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
log.info("提交延迟任务");
// 发送消息
rabbitTemplate.convertAndSend("sanyouDirectExchangee", "", task, correlationData);
}
}
整个工作流程如下:<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<version>2.2.5.RELEASE</version>
</dependency>spring:
redis:
host: 192.168.200.144
port: 6379@Configuration
public class RedisConfiguration {
@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) {
RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
redisMessageListenerContainer.setConnectionFactory(connectionFactory);
return redisMessageListenerContainer;
}
@Bean
public KeyExpirationEventMessageListener redisKeyExpirationListener(RedisMessageListenerContainer redisMessageListenerContainer) {
return new KeyExpirationEventMessageListener(redisMessageListenerContainer);
}
}@Component
public class MyRedisKeyExpiredEventListener implements ApplicationListener<RedisKeyExpiredEvent> {
@Override
public void onApplicationEvent(RedisKeyExpiredEvent event) {
byte[] body = event.getSource();
System.out.println("获取到延迟消息:" + new String(body));
}
}set sanyou task
expire sanyou 5
虽然这种方式可以实现延迟任务,但是这种方式坑比较多
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.1</version>
</dependency>@Component
@Slf4j
public class RedissonDelayQueue {
private RedissonClient redissonClient;
private RDelayedQueue<String> delayQueue;
private RBlockingQueue<String> blockingQueue;
@PostConstruct
public void init() {
initDelayQueue();
startDelayQueueConsumer();
}
private void initDelayQueue() {
Config config = new Config();
SingleServerConfig serverConfig = config.useSingleServer();
serverConfig.setAddress("redis://localhost:6379");
redissonClient = Redisson.create(config);
blockingQueue = redissonClient.getBlockingQueue("SANYOU");
delayQueue = redissonClient.getDelayedQueue(blockingQueue);
}
private void startDelayQueueConsumer() {
new Thread(() -> {
while (true) {
try {
String task = blockingQueue.take();
log.info("接收到延迟任务:{}", task);
} catch (Exception e) {
e.printStackTrace();
}
}
}, "SANYOU-Consumer").start();
}
public void offerTask(String task, long seconds) {
log.info("添加延迟任务:{} 延迟时间:{}s", task, seconds);
delayQueue.offer(task, seconds, TimeUnit.SECONDS);
}
}@RestController
public class RedissonDelayQueueController {
@Resource
private RedissonDelayQueue redissonDelayQueue;
@GetMapping("/add")
public void addTask(@RequestParam("task") String task) {
redissonDelayQueue.offerTask(task, 5);
}
}@Slf4j
public class NettyHashedWheelTimerDemo {
public static void main(String[] args) {
HashedWheelTimer timer = new HashedWheelTimer(100, TimeUnit.MILLISECONDS, 8);
timer.start();
log.info("提交延迟任务");
timer.newTimeout(timeout -> log.info("执行延迟任务"), 5, TimeUnit.SECONDS);
}
}@Slf4j
public class SystemTimerDemo {
public static void main(String[] args) {
SystemTimer systemTimer = new SystemTimer();
systemTimer.start();
log.info("提交延迟任务");
systemTimer.addTask(new TimerTask(() -> log.info("执行延迟任务"), 5000));
}
}<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.2</version>
</dependency>@Slf4j
public class SanYouJob implements Job {
@Override
public void execute(JobExecutionContext context) throws JobExecutionException {
JobDetail jobDetail = context.getJobDetail();
JobDataMap jobDataMap = jobDetail.getJobDataMap();
log.info("获取到延迟任务:{}", jobDataMap.get("delayTask"));
}
}public class QuartzDemo {
public static void main(String[] args) throws SchedulerException, InterruptedException {
// 1.创建Scheduler的工厂
SchedulerFactory sf = new StdSchedulerFactory();
// 2.从工厂中获取调度器实例
Scheduler scheduler = sf.getScheduler();
// 6.启动 调度器
scheduler.start();
// 3.创建JobDetail,Job类型就是上面说的SanYouJob
JobDetail jb = JobBuilder.newJob(SanYouJob.class)
.usingJobData("delayTask", "这是一个延迟任务")
.build();
// 4.创建Trigger
Trigger t = TriggerBuilder.newTrigger()
//任务的触发时间就是延迟任务到的延迟时间
.startAt(DateUtil.offsetSecond(new Date(), 5))
.build();
// 5.注册任务和定时器
log.info("提交延迟任务");
scheduler.scheduleJob(jb, t);
}
}
启动的时候会开启一个QuartzSchedulerThread调度线程,这个线程会去判断任务是否到了执行时间,到的话就将任务交给任务线程池去执行。@Slf4j
public class PollingTaskDemo {
private static final List<DelayTask> DELAY_TASK_LIST = new CopyOnWriteArrayList<>();
public static void main(String[] args) {
new Thread(() -> {
while (true) {
try {
for (DelayTask delayTask : DELAY_TASK_LIST) {
if (delayTask.triggerTime <= System.currentTimeMillis()) {
log.info("处理延迟任务:{}", delayTask.taskContent);
DELAY_TASK_LIST.remove(delayTask);
}
}
TimeUnit.MILLISECONDS.sleep(100);
} catch (Exception e) {
}
}
}).start();
log.info("提交延迟任务");
DELAY_TASK_LIST.add(new DelayTask("三友的java日记", 5L));
}
@Getter
@Setter
public static class DelayTask {
private final String taskContent;
private final Long triggerTime;
public DelayTask(String taskContent, Long delayTime) {
this.taskContent = taskContent;
this.triggerTime = System.currentTimeMillis() + delayTime * 1000;
}
}
}
这种操作简单,但是就是效率低下,每次都得遍历所有的任务。我试图在一个项目中使用rake,如果我把所有东西都放到Rakefile中,它会很大并且很难读取/找到东西,所以我试着将每个命名空间放在lib/rake中它自己的文件中,我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题,但没有任务。我现在只有一个.rake文件作为测试,名为“servers.rake”,它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时
我试图获取一个长度在1到10之间的字符串,并输出将字符串分解为大小为1、2或3的连续子字符串的所有可能方式。例如:输入:123456将整数分割成单个字符,然后继续查找组合。该代码将返回以下所有数组。[1,2,3,4,5,6][12,3,4,5,6][1,23,4,5,6][1,2,34,5,6][1,2,3,45,6][1,2,3,4,56][12,34,5,6][12,3,45,6][12,3,4,56][1,23,45,6][1,2,34,56][1,23,4,56][12,34,56][123,4,5,6][1,234,5,6][1,2,345,6][1,2,3,456][123
我主要使用Ruby来执行此操作,但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式,例如使用这个yaml文件,它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
如何使用RSpec::Core::RakeTask初始化RSpecRake任务?require'rspec/core/rake_task'RSpec::Core::RakeTask.newdo|t|#whatdoIputinhere?endInitialize函数记录在http://rubydoc.info/github/rspec/rspec-core/RSpec/Core/RakeTask#initialize-instance_method没有很好的记录;它只是说:-(RakeTask)initialize(*args,&task_block)AnewinstanceofRake
我有一个用户工厂。我希望默认情况下确认用户。但是鉴于unconfirmed特征,我不希望它们被确认。虽然我有一个基于实现细节而不是抽象的工作实现,但我想知道如何正确地做到这一点。factory:userdoafter(:create)do|user,evaluator|#unwantedimplementationdetailshereunlessFactoryGirl.factories[:user].defined_traits.map(&:name).include?(:unconfirmed)user.confirm!endendtrait:unconfirmeddoenden
question的一些答案关于redirect_to让我想到了其他一些问题。基本上,我正在使用Rails2.1编写博客应用程序。我一直在尝试自己完成大部分工作(因为我对Rails有所了解),但在需要时会引用Internet上的教程和引用资料。我设法让一个简单的博客正常运行,然后我尝试添加评论。靠我自己,我设法让它进入了可以从script/console添加评论的阶段,但我无法让表单正常工作。我遵循的其中一个教程建议在帖子Controller中创建一个“评论”操作,以添加评论。我的问题是:这是“标准”方式吗?我的另一个问题的答案之一似乎暗示应该有一个CommentsController参
华为OD机试题本篇题目:明明的随机数题目输入描述输出描述:示例1输入输出说明代码编写思路最近更新的博客华为od2023|什么是华为od,od薪资待遇,od机试题清单华为OD机试真题大全,用Python解华为机试题|机试宝典【华为OD机试】全流程解析+经验分享,题型分享,防作弊指南华为o
在应用开发中,有时候我们需要获取系统的设备信息,用于数据上报和行为分析。那在鸿蒙系统中,我们应该怎么去获取设备的系统信息呢,比如说获取手机的系统版本号、手机的制造商、手机型号等数据。1、获取方式这里分为两种情况,一种是设备信息的获取,一种是系统信息的获取。1.1、获取设备信息获取设备信息,鸿蒙的SDK包为我们提供了DeviceInfo类,通过该类的一些静态方法,可以获取设备信息,DeviceInfo类的包路径为:ohos.system.DeviceInfo.具体的方法如下:ModifierandTypeMethodDescriptionstatic StringgetAbiList()Obt
C#实现简易绘图工具一.引言实验目的:通过制作窗体应用程序(C#画图软件),熟悉基本的窗体设计过程以及控件设计,事件处理等,熟悉使用C#的winform窗体进行绘图的基本步骤,对于面向对象编程有更加深刻的体会.Tutorial任务设计一个具有基本功能的画图软件**·包括简单的新建文件,保存,重新绘图等功能**·实现一些基本图形的绘制,包括铅笔和基本形状等,学习橡皮工具的创建**·设计一个合理舒适的UI界面**注明:你可能需要先了解一些关于winform窗体应用程序绘图的基本知识,以及关于GDI+类和结构的知识二.实验环境Windows系统下的visualstudio2017C#窗体应用程序三.
运行bundleinstall后出现此错误:Gem::Package::FormatError:nometadatafoundin/Users/jeanosorio/.rvm/gems/ruby-1.9.3-p286/cache/libv8-3.11.8.13-x86_64-darwin-12.gemAnerroroccurredwhileinstallinglibv8(3.11.8.13),andBundlercannotcontinue.Makesurethat`geminstalllibv8-v'3.11.8.13'`succeedsbeforebundling.我试试gemin