Jesque 基础入门

Jesque是Resque的java实现，它是一个maven项目，依赖Jedis连接到Redis，使用Jackson序列化Job信息，存储在Redis数据库中。主要用于做定时任务。

添加依赖

你可以去官网下载jar包，当然如果你是一个maven项目，像下面这样直接添加依赖即可：

<dependency>
  <groupId>net.greghaines</groupId>
  <artifactId>jesque</artifactId>
  <version>2.0.2</version>
</dependency>

创建即时任务

final Job job = new Job("TestAction", 
    new Object[]{ 1, 2.3, true, "test", Arrays.asList("inner", 4.5)});

client.enqueue("queue", job);

创建定时任务

final Job job = new Job("TestAction", 
    new Object[]{ 1, 2.3, true, "test", Arrays.asList("inner", 4.5)});

final long delay = 10; // 秒
final long future = System.currentTimeMillis() + 
    (delay * 1000); // job执行时间点，格式：时间戳，单位：毫秒

// 10秒后执行job
client.delayedEnqueue("delayQueue", job, future);

创建循环定时任务

final Job job = new Job("TestAction", 
    new Object[]{ 1, 2.3, true, "test", Arrays.asList("inner", 4.5)});

final long delay = 10; // 秒
final long future = System.currentTimeMillis() + 
    (delay * 1000); // job执行时间点，格式：时间戳，单位：毫秒
final long frequency = 60; // 秒

// 10秒后执行job，以后每隔1分钟执行一次
client.recurringEnqueue("recurDelayQueue", job, future, (frequency * 1000));

取消定时任务

// 取消redis中名为delayQueue的set集合存放的job
client.removeDelayedEnqueue("delayQueue", job);

client.removeRecurringEnqueue("recurDelayQueue", job);

监听消费者

在这里我只是把worker命令为消费者，仅仅是因为它看起来像是在消费我们创建的Jesque任务而已。

worker.getWorkerEventEmitter().addListener(new WorkerListener(){
   public void onEvent(WorkerEvent event, Worker worker, String queue, 
       Job job, Object runner, Object result, Throwable t) {
        if (runner instanceof TestAction) {
            // do something
        }
    }
}, WorkerEvent.JOB_EXECUTE);

我们可以监听到的消费事件，你可以监听这些事件在相应时刻做一些事情：

事件	描述
WORKER_START	完成启动,即将开始运行。
WORKER_POLL	轮询队列。
JOB_PROCESS	处理任务。
JOB_EXECUTE	执行任务。
JOB_SUCCESS	任务执行成功。
JOB_FAILURE	任务执行失败。
WORKER_ERROR	发现在正常运行时产生异常。
WORKER_STOP	完成运行，即将关闭。

增加Jedis配置

{
    InputStream inputStream = JesqueConfig.class.getResourceAsStream(
        "/db.properties");
    Properties properties = new Properties();
    try {
        properties.load(inputStream);
    } catch (IOException e) {
        e.printStackTrace();
    }

    String host = properties.getProperty("redis.host");
    String port = properties.getProperty("redis.port");
    ConfigBuilder configBuilder = new ConfigBuilder();
    if (!StringUtils.isEmpty(host)) {
        configBuilder.withHost(host);
    }
    if (!StringUtils.isEmpty(port)) {
        configBuilder.withPort(Integer.parseInt(port));
    }

    setRedisConfig(configBuilder.build());
}

参考链接：Jesque官网 & 我的一个简单示例