public class MyJob implements Job {
public void execute(JobExecutionContext jobExecutionContext)
throws JobExecutionException {
System.out.println("进入数据库");
}
public static void main(String[] args) throws SchedulerException {
// 创建任务
JobDetail jobDetail = JobBuilder.newJob(MyJob.class)
.withIdentity("job1","group1").build();
// 创建触发器Trigger 每三秒执行一次
Trigger trigger= TriggerBuilder
.newTrigger()
.withIdentity("trigger1", "group3")
.withSchedule(
SimpleScheduleBuilder
.simpleSchedule()
.withIntervalInSeconds(3)
.repeatForever()
) .build();
Scheduler scheduler = new StdSchedulerFactory().getScheduler();
// 将任务及其触发器放入调度器
scheduler.scheduleJob(jobDetail,trigger);
// 调度器开始调度任务
scheduler.start();
}
}
2、延迟队列
public class OrderDelay implements Delayed {
private String orderId;
private Long timeout;
OrderDelay(String orderId, Long timeout) {
this.orderId = orderId;
this.timeout = timeout + System.nanoTime();
}
public int compareTo(Delayed other) {
if(other == this) {
return 0;
}
OrderDelay t = (OrderDelay) other;
Long d = (getDelay(TimeUnit.NANOSECONDS) - t
.getDelay(TimeUnit.NANOSECONDS));
return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
}
public long getDelay(TimeUnit unit) {
return unit.convert(
timeout - System.nanoTime(),
TimeUnit.NANOSECONDS);
}
void print() {
System.out.println(orderId + "开始启动了");
}
}
3、时间轮算法
package wheeltime;
import io.netty.util.*;
import io.netty.util.TimerTask;
import java.util.concurrent.*;
public class HashedWheelTimerTest {
static class MyTimerTask implements TimerTask {
boolean flag;
public MyTimerTask(boolean flag) {
this.flag = flag;
}
public void run(Timeout timeout) throws Exception {
System.out.println("要去删除了。。。");
this.flag = false;
}
}
public static void main(String[] argv) {
MyTimerTask timerTask = new MyTimerTask(true);
Timer timer = new HashedWheelTimer();
//轮数，时间，时间单位
timer.newTimeout(timerTask, 5, TimeUnit.SECONDS);
int i = 1;
while(timerTask.flag) {
try {
Thread.sleep(1000);
} catch(InterruptedException e) {
e.printStackTrace();
}
System.out.println(i + "过去了");
i++;
}
}
}
4、Redis缓存
方式一：
利用redis的zset,zset是一个有序集合，每一个元素(member)都关联了一个score,通过score排序来取集合中的值
利用redis命令理解思路 添加单个元素 redis> ZADD page_rank 10 google.com (integer) 1 添加多个元素 redis> ZADD page_rank 9 baidu.com 8 bing.com (integer) 2 redis> ZRANGE page_rank 0 -1 WITHSCORES 1) "bing.com" 2) "8" 3) "baidu.com" 4) "9" 5) "google.com" 6) "10" 查询元素的score值 redis> ZSCORE page_rank bing.com "8" 移除单个元素 redis> ZREM page_rank google.com (integer) 1 redis> ZRANGE page_rank 0 -1 WITHSCORES 1) "bing.com" 2) "8" 3) "baidu.com" 4) "9"
java实现：
package timestop.redisout;
import redis.clients.jedis.*;
import java.util.*;
public class AppTest {
private static final String ADDR = "127.0.0.1";
private static final int PORT = 6379;
//JedisPool 创建线程安全的网络连接池
private static JedisPool JedisPool = new JedisPool(ADDR, PORT);
//获取连接池的一个jedis对象
public static Jedis getJedis() {
return JedisPool.getResource();
}
// 生产者， 生成5个订单放进去
public void productionDelayMessage() {
for(int i = 0; i < 5; i++) {
Calendar cal1 = Calendar.getInstance();
//Calendar.SECOND时间单位，amount时间数
cal1.add(Calendar.SECOND, 3);
//cal1.getTimeInMillis()用于返回此日历的当前时间 (以毫秒为单位)。
int second3later = (int) (cal1.getTimeInMillis() / 1000);
//getJedis().zadd添加元素
AppTest.getJedis().zadd("OrderId", second3later, "OID0000001" + i);
System.out.println(System.currentTimeMillis() + "ms:redis生成了一个订单任务：订单ID为"+"OID0000001" + i);
}
}
// 消费者, 取订单
public void consumerDelayMessage() {
Jedis jedis = AppTest.getJedis();
while(true) {
//返回有序集合中指定分数区间的成员列表。
Set items = jedis.zrangeWithScores("OrderId", 0, 1);
if(items == null || items.isEmpty()) {
System.out.println("当前没有等待的任务");
try {
Thread.sleep(500);
} catch(InterruptedException e) {
e.printStackTrace();
}
continue;
}
int score = (int) ((Tuple) items.toArray()[0]).getScore();
Calendar cal = Calendar.getInstance();
//cal1.getTimeInMillis()用于返回此日历的当前时间 (以毫秒为单位)。
int nowSecond = (int) (cal.getTimeInMillis() / 1000);
//原版 有多个线程消费同一个资源的情况
//if(nowSecond >= score) {
// String orderId = ((Tuple) items.toArray()[0]).getElement();
// jedis.zrem("OrderId", orderId);
// System.out.println(System.currentTimeMillis() + "ms:redis消费了一个任务：消费的订单OrderId为" + orderId);
//}
//改进后
if(nowSecond >= score) {
String orderId = ((Tuple) items.toArray()[0]).getElement();
Long num = jedis.zrem("OrderId", orderId);
if(num != null && num > 0) {
System.out.println(System.currentTimeMillis() + "ms:redis消费了一个任务：消费的订单OrderId为" + orderId);
}
}
}
}
public static void main(String[] args) {
AppTest appTest = new AppTest();
appTest.productionDelayMessage();
appTest.consumerDelayMessage();
}
}
方式二：
该方案使用redis的Keyspace Notifications，中文翻译就是键空间机制，就是利用该机制可以在key失效之后，提供一个回调，实际上是redis会给客户端发送一个消息。是需要redis版本2.8以上。 实现二 在redis.conf中，加入一条配置 notify-keyspace-events Ex
public class RedisTest {
private static final String ADDR = "127.0.0.1";
private static final int PORT = 6379;
private static JedisPool jedis = new JedisPool(ADDR, PORT);
private static RedisSub sub = new RedisSub();
public static void init() {
new Thread(new Runnable() {
public void run() {
jedis.getResource().subscribe(sub, "__keyevent@0__:expired");
}
}).start();
}
public static void main(String[] args) throws InterruptedException {
init();
for(int i =0;i<10;i++){
String orderId = "OID000000"+i;
jedis.getResource().setex(orderId, 3, orderId);
System.out.println(System.currentTimeMillis()+"ms:"+orderId+"订单生成");
}
}
static class RedisSub extends JedisPubSub {
@Override
public void onMessage(String channel, String message) {
System.out.println(System.currentTimeMillis()+"ms:"+message+"订单取消");
}
}
}
