From c3a88eae48b556059366cb837741cac5e9a0d801 Mon Sep 17 00:00:00 2001 From: ofirnk Date: Sun, 27 Mar 2016 23:55:46 +0300 Subject: [PATCH 1/4] MultiplePriorityQueues: Add ability to POP first message from a list of queues --- .../net/greghaines/jesque/worker/Worker.java | 7 + .../greghaines/jesque/worker/WorkerImpl.java | 163 ++++++++++++++---- .../greghaines/jesque/worker/WorkerPool.java | 8 + .../fromMultiplePriorityQueues.lua | 34 ++++ .../jesque/worker/TestWorkerImpl.java | 41 ++++- .../jesque/worker/TestWorkerPool.java | 34 ++-- 6 files changed, 236 insertions(+), 51 deletions(-) create mode 100644 src/main/resources/workerScripts/fromMultiplePriorityQueues.lua diff --git a/src/main/java/net/greghaines/jesque/worker/Worker.java b/src/main/java/net/greghaines/jesque/worker/Worker.java index 497605ea..a8f0413e 100644 --- a/src/main/java/net/greghaines/jesque/worker/Worker.java +++ b/src/main/java/net/greghaines/jesque/worker/Worker.java @@ -18,6 +18,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.List; /** * A Worker polls for Jobs from a specified list of queues, executing them in @@ -93,6 +94,12 @@ public interface Worker extends JobExecutor, Runnable { * @param queues the queues to poll */ void setQueues(Collection queues); + + /** + * Clear any current queues and poll the given queues in the specified order. + * @param queues the queues to poll + */ + void setQueues(List queues); /** * @return the worker event emitter for this worker diff --git a/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java b/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java index b9fa844e..cdd55173 100644 --- a/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java +++ b/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java @@ -15,19 +15,32 @@ */ package net.greghaines.jesque.worker; -import static net.greghaines.jesque.utils.ResqueConstants.*; -import static net.greghaines.jesque.worker.JobExecutor.State.*; -import static net.greghaines.jesque.worker.WorkerEvent.*; +import com.fasterxml.jackson.core.JsonParseException; +import com.fasterxml.jackson.databind.JsonMappingException; +import net.greghaines.jesque.Config; +import net.greghaines.jesque.Job; +import net.greghaines.jesque.JobFailure; +import net.greghaines.jesque.WorkerStatus; +import net.greghaines.jesque.json.ObjectMapperFactory; +import net.greghaines.jesque.utils.JedisUtils; +import net.greghaines.jesque.utils.JesqueUtils; +import net.greghaines.jesque.utils.ScriptUtils; +import net.greghaines.jesque.utils.VersionUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.exceptions.JedisException; import java.io.IOException; -import java.lang.Throwable; import java.lang.management.ManagementFactory; import java.net.InetAddress; import java.net.UnknownHostException; import java.text.SimpleDateFormat; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Date; +import java.util.List; import java.util.concurrent.BlockingDeque; import java.util.concurrent.Callable; import java.util.concurrent.LinkedBlockingDeque; @@ -36,24 +49,32 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; -import net.greghaines.jesque.Config; -import net.greghaines.jesque.Job; -import net.greghaines.jesque.JobFailure; -import net.greghaines.jesque.WorkerStatus; -import net.greghaines.jesque.json.ObjectMapperFactory; -import net.greghaines.jesque.utils.JedisUtils; -import net.greghaines.jesque.utils.JesqueUtils; -import net.greghaines.jesque.utils.ScriptUtils; -import net.greghaines.jesque.utils.VersionUtils; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import redis.clients.jedis.Jedis; -import redis.clients.jedis.exceptions.JedisException; - -import com.fasterxml.jackson.core.JsonParseException; -import com.fasterxml.jackson.databind.JsonMappingException; +import static net.greghaines.jesque.utils.ResqueConstants.COLON; +import static net.greghaines.jesque.utils.ResqueConstants.DATE_FORMAT; +import static net.greghaines.jesque.utils.ResqueConstants.FAILED; +import static net.greghaines.jesque.utils.ResqueConstants.INFLIGHT; +import static net.greghaines.jesque.utils.ResqueConstants.JAVA_DYNAMIC_QUEUES; +import static net.greghaines.jesque.utils.ResqueConstants.PROCESSED; +import static net.greghaines.jesque.utils.ResqueConstants.QUEUE; +import static net.greghaines.jesque.utils.ResqueConstants.QUEUES; +import static net.greghaines.jesque.utils.ResqueConstants.STARTED; +import static net.greghaines.jesque.utils.ResqueConstants.STAT; +import static net.greghaines.jesque.utils.ResqueConstants.WORKER; +import static net.greghaines.jesque.utils.ResqueConstants.WORKERS; +import static net.greghaines.jesque.worker.JobExecutor.State.NEW; +import static net.greghaines.jesque.worker.JobExecutor.State.RUNNING; +import static net.greghaines.jesque.worker.JobExecutor.State.SHUTDOWN; +import static net.greghaines.jesque.worker.JobExecutor.State.SHUTDOWN_IMMEDIATE; +import static net.greghaines.jesque.worker.WorkerEvent.JOB_EXECUTE; +import static net.greghaines.jesque.worker.WorkerEvent.JOB_FAILURE; +import static net.greghaines.jesque.worker.WorkerEvent.JOB_PROCESS; +import static net.greghaines.jesque.worker.WorkerEvent.JOB_SUCCESS; +import static net.greghaines.jesque.worker.WorkerEvent.WORKER_ERROR; +import static net.greghaines.jesque.worker.WorkerEvent.WORKER_POLL; +import static net.greghaines.jesque.worker.WorkerEvent.WORKER_START; +import static net.greghaines.jesque.worker.WorkerEvent.WORKER_STOP; +import static net.greghaines.jesque.worker.WorkerImpl.NextQueueStrategy.DRAIN_WHILE_MESSAGES_EXISTS; +import static net.greghaines.jesque.worker.WorkerImpl.NextQueueStrategy.RESET_TO_HIGHEST_PRIORITY; /** * WorkerImpl is an implementation of the Worker interface. Obeys the contract of a Resque worker in Redis. @@ -65,9 +86,12 @@ public class WorkerImpl implements Worker { protected static final long EMPTY_QUEUE_SLEEP_TIME = 500; // 500 ms protected static final long RECONNECT_SLEEP_TIME = 5000; // 5 sec protected static final int RECONNECT_ATTEMPTS = 120; // Total time: 10 min - + private static final String LPOPLPUSH_LUA = "/workerScripts/jesque_lpoplpush.lua"; + private static final String POP_LUA = "/workerScripts/jesque_pop.lua"; + // Set the thread name to the message for debugging private static volatile boolean threadNameChangingEnabled = false; + private final NextQueueStrategy nextQueueStrategy; /** * @return true if worker threads names will change during normal operation @@ -113,6 +137,7 @@ protected static void checkQueues(final Iterable queues) { private final AtomicBoolean processingJob = new AtomicBoolean(false); private final AtomicReference popScriptHash = new AtomicReference<>(null); private final AtomicReference lpoplpushScriptHash = new AtomicReference<>(null); + private final AtomicReference multiPriorityQueuesScriptHash = new AtomicReference<>(null); private final long workerId = WORKER_COUNTER.getAndIncrement(); private final String threadNameBase = "Worker-" + this.workerId + " Jesque-" + VersionUtils.getVersion() + ": "; private final AtomicReference threadRef = new AtomicReference(null); @@ -144,6 +169,25 @@ public WorkerImpl(final Config config, final Collection queues, final Jo */ public WorkerImpl(final Config config, final Collection queues, final JobFactory jobFactory, final Jedis jedis) { + this(config, + (queues == null ? Collections.EMPTY_LIST : new ArrayList<>(queues)), + jobFactory, + jedis, + DRAIN_WHILE_MESSAGES_EXISTS); + } + + /** + * Creates a new WorkerImpl, with the given connection to Redis.
+ * The worker will only listen to the supplied queues and execute jobs that are provided by the given job factory. + * @param config used to create a connection to Redis and the package prefix for incoming jobs + * @param queues the list of queues to poll + * @param jobFactory the job factory that materializes the jobs + * @param jedis the connection to Redis + * @param nextQueueStrategy defines worker behaviour once it has found messages in a queue + * @throws IllegalArgumentException if either config, queues, jobFactory or jedis is null + */ + public WorkerImpl(final Config config, final List queues, final JobFactory jobFactory, + final Jedis jedis, NextQueueStrategy nextQueueStrategy) { if (config == null) { throw new IllegalArgumentException("config must not be null"); } @@ -154,6 +198,7 @@ public WorkerImpl(final Config config, final Collection queues, final Jo throw new IllegalArgumentException("jedis must not be null"); } checkQueues(queues); + this.nextQueueStrategy = nextQueueStrategy; this.config = config; this.jobFactory = jobFactory; this.namespace = config.getNamespace(); @@ -185,9 +230,9 @@ public void run() { this.jedis.sadd(key(WORKERS), this.name); this.jedis.set(key(WORKER, this.name, STARTED), new SimpleDateFormat(DATE_FORMAT).format(new Date())); this.listenerDelegate.fireEvent(WORKER_START, this, null, null, null, null, null); - this.popScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript("/workerScripts/jesque_pop.lua"))); - this.lpoplpushScriptHash.set(this.jedis.scriptLoad( - ScriptUtils.readScript("/workerScripts/jesque_lpoplpush.lua"))); + this.popScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(POP_LUA))); + this.lpoplpushScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(LPOPLPUSH_LUA))); + this.multiPriorityQueuesScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(LPOPLPUSH_LUA))); poll(); } catch (Exception ex) { LOG.error("Uncaught exception in worker run-loop!", ex); @@ -328,6 +373,14 @@ public void removeAllQueues() { */ @Override public void setQueues(final Collection queues) { + setQueues(new ArrayList<>(queues)); + } + + /** + * {@inheritDoc} + */ + @Override + public void setQueues(final List queues) { checkQueues(queues); this.queueNames.clear(); this.queueNames.addAll((queues == ALL_QUEUES) // Using object equality on purpose @@ -408,9 +461,8 @@ protected void poll() { if (threadNameChangingEnabled) { renameThread("Waiting for " + JesqueUtils.join(",", this.queueNames)); } - curQueue = this.queueNames.poll(EMPTY_QUEUE_SLEEP_TIME, TimeUnit.MILLISECONDS); + curQueue = getNextQueue(); if (curQueue != null) { - this.queueNames.add(curQueue); // Rotate the queues checkPaused(); // Might have been waiting in poll()/checkPaused() for a while if (RUNNING.equals(this.state.get())) { @@ -419,10 +471,13 @@ protected void poll() { if (payload != null) { process(ObjectMapperFactory.get().readValue(payload, Job.class), curQueue); missCount = 0; - } else if (++missCount >= this.queueNames.size() && RUNNING.equals(this.state.get())) { - // Keeps worker from busy-spinning on empty queues - missCount = 0; - Thread.sleep(EMPTY_QUEUE_SLEEP_TIME); + } else { + missCount++; + if (shouldSleep(missCount) && RUNNING.equals(this.state.get())) { + // Keeps worker from busy-spinning on empty queues + missCount = 0; + Thread.sleep(EMPTY_QUEUE_SLEEP_TIME); + } } } } @@ -440,6 +495,27 @@ protected void poll() { } } + private boolean shouldSleep(int missCount) { + return nextQueueStrategy == RESET_TO_HIGHEST_PRIORITY || + missCount >= this.queueNames.size(); + } + + protected String getNextQueue() throws InterruptedException { + switch (nextQueueStrategy) { + case DRAIN_WHILE_MESSAGES_EXISTS: + final String nextPollQueue = this.queueNames.poll(EMPTY_QUEUE_SLEEP_TIME, TimeUnit.MILLISECONDS); + if (nextPollQueue != null) { + // Rotate the queues + this.queueNames.add(nextPollQueue); + } + return nextPollQueue; + case RESET_TO_HIGHEST_PRIORITY: + return JesqueUtils.join(",", this.queueNames); + default: + throw new RuntimeException("Unimplemented 'nextQueueStrategy'"); + } + } + /** * Remove a job from the given queue. * @param curQueue the queue to remove a job from @@ -447,8 +523,15 @@ protected void poll() { */ protected String pop(final String curQueue) { final String key = key(QUEUE, curQueue); - return (String) this.jedis.evalsha(this.popScriptHash.get(), 3, key, key(INFLIGHT, this.name, curQueue), - JesqueUtils.createRecurringHashKey(key), Long.toString(System.currentTimeMillis())); + switch (nextQueueStrategy) { + case DRAIN_WHILE_MESSAGES_EXISTS: + return (String) this.jedis.evalsha(this.popScriptHash.get(), 3, key, key(INFLIGHT, this.name, curQueue), + JesqueUtils.createRecurringHashKey(key), Long.toString(System.currentTimeMillis())); + case RESET_TO_HIGHEST_PRIORITY: + return (String) this.jedis.evalsha(this.multiPriorityQueuesScriptHash.get(), 1, curQueue); + default: + throw new RuntimeException("Unimplemented 'nextQueueStrategy'"); + } } /** @@ -708,4 +791,16 @@ protected String lpoplpush(final String from, final String to) { public String toString() { return this.namespace + COLON + WORKER + COLON + this.name; } + + public enum NextQueueStrategy { + /** + * Would drain messages as long as current queue is not empty + */ + DRAIN_WHILE_MESSAGES_EXISTS, + + /** + * Would reset to check {first queue, second queue, etc'} after each message + */ + RESET_TO_HIGHEST_PRIORITY + } } diff --git a/src/main/java/net/greghaines/jesque/worker/WorkerPool.java b/src/main/java/net/greghaines/jesque/worker/WorkerPool.java index 371af61b..f9fe327c 100644 --- a/src/main/java/net/greghaines/jesque/worker/WorkerPool.java +++ b/src/main/java/net/greghaines/jesque/worker/WorkerPool.java @@ -253,6 +253,14 @@ public void removeAllQueues() { */ @Override public void setQueues(final Collection queues) { + setQueues(new ArrayList<>(queues)); + } + + /** + * {@inheritDoc} + */ + @Override + public void setQueues(final List queues) { for (final Worker worker : this.workers) { worker.setQueues(queues); } diff --git a/src/main/resources/workerScripts/fromMultiplePriorityQueues.lua b/src/main/resources/workerScripts/fromMultiplePriorityQueues.lua new file mode 100644 index 00000000..1cbc15b0 --- /dev/null +++ b/src/main/resources/workerScripts/fromMultiplePriorityQueues.lua @@ -0,0 +1,34 @@ +-- +-- User: Ofir Naor +-- Date: 3/27/16 +-- Time: 2:47 PM +-- +local queues = KEYS[1] +--local debug = {'Hooray!', "'"..queues.."'"} + +local QUEUE_NAME_CAPTURING_REGEX = '([^,]+)' +local OPTIONAL_COMMA_SEPARATOR = ',?' +local OPTIONAL_SPACE_SEPARATOR = '%s*' +local NEXT_QUEUE_REGEX = QUEUE_NAME_CAPTURING_REGEX .. OPTIONAL_COMMA_SEPARATOR .. OPTIONAL_SPACE_SEPARATOR + +for q in queues.gmatch(queues, NEXT_QUEUE_REGEX) do + local firstMsg = redis.call('ZRANGE', q, '0', '0') +-- table.insert(debug, "q: " .. q) + if firstMsg ~= nil then +-- table.insert(debug, 'result:') +-- table.insert(debug, firstMsg) +-- table.insert(debug, 'removed:') + local payload = firstMsg[1] + if payload ~= nil then + local removedItems = redis.call('ZREM', q, payload) +-- table.insert(debug, payload) +-- table.insert(debug, "num of removed items:") +-- table.insert(debug, type(payload)) + return payload +-- return debug + end + end +end +--return debug +--return "NOT_FOUND" +return nil \ No newline at end of file diff --git a/src/test/java/net/greghaines/jesque/worker/TestWorkerImpl.java b/src/test/java/net/greghaines/jesque/worker/TestWorkerImpl.java index fe05d80d..f5f745cd 100644 --- a/src/test/java/net/greghaines/jesque/worker/TestWorkerImpl.java +++ b/src/test/java/net/greghaines/jesque/worker/TestWorkerImpl.java @@ -1,21 +1,27 @@ package net.greghaines.jesque.worker; -import static net.greghaines.jesque.utils.JesqueUtils.entry; -import static net.greghaines.jesque.utils.JesqueUtils.map; - -import java.util.Arrays; - import net.greghaines.jesque.Config; import net.greghaines.jesque.ConfigBuilder; import net.greghaines.jesque.TestAction; - +import org.jmock.Mockery; +import org.jmock.integration.junit4.JUnit4Mockery; +import org.jmock.internal.InvocationExpectation; +import org.jmock.lib.legacy.ClassImposteriser; import org.junit.Assert; import org.junit.Test; +import redis.clients.jedis.Jedis; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; + +import static net.greghaines.jesque.utils.JesqueUtils.entry; +import static net.greghaines.jesque.utils.JesqueUtils.map; public class TestWorkerImpl { private static final Config CONFIG = new ConfigBuilder().build(); - + @Test(expected=IllegalArgumentException.class) public void testConstructor_NullConfig() { new WorkerImpl(null, null, null, null); @@ -58,4 +64,25 @@ public void testCheckQueues_EmptyQueue() { public void testCheckQueues_OK() { WorkerImpl.checkQueues(Arrays.asList("foo", "bar")); } + + @Test + public void verifyNoExceptionsForAllNextQueueStrategies() throws InterruptedException { + final MapBasedJobFactory jobFactory = new MapBasedJobFactory(Collections.EMPTY_MAP); + for (WorkerImpl.NextQueueStrategy nextQueueStrategy : WorkerImpl.NextQueueStrategy.values()) { + final WorkerImpl worker = new WorkerImpl(CONFIG, + new ArrayList(), + jobFactory, + getJedis(), + nextQueueStrategy); + worker.pop(worker.getNextQueue()); + } + } + + private Jedis getJedis() { + final Mockery mockCtx = new JUnit4Mockery(); + mockCtx.setImposteriser(ClassImposteriser.INSTANCE); + final Jedis jedis = mockCtx.mock(Jedis.class); + mockCtx.addExpectation(new InvocationExpectation()); + return jedis; + } } diff --git a/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java b/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java index 07f3f2ae..bbce8004 100644 --- a/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java +++ b/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java @@ -1,13 +1,5 @@ package net.greghaines.jesque.worker; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.Callable; - import org.jmock.Expectations; import org.jmock.Mockery; import org.jmock.integration.junit4.JUnit4Mockery; @@ -17,6 +9,15 @@ import org.junit.Before; import org.junit.Test; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Callable; + public class TestWorkerPool { private static final int NUM_WORKERS = 2; @@ -214,14 +215,27 @@ public void testRemoveAllQueues() { @Test public void testSetQueues() { - final Collection queues = Arrays.asList("queue1", "queue2"); + final List queues = Arrays.asList("queue1", "queue2"); this.mockCtx.checking(new Expectations(){{ oneOf(workers.get(0)).setQueues(queues); oneOf(workers.get(1)).setQueues(queues); }}); this.pool.setQueues(queues); } - + + @Test + public void testSetCollectionQueues() { + final Collection queues = new HashSet<>(); + queues.add("queue1"); + queues.add("queue2"); + final List queuesAsList = Arrays.asList("queue1", "queue2"); + this.mockCtx.checking(new Expectations(){{ + oneOf(workers.get(0)).setQueues(queuesAsList); + oneOf(workers.get(1)).setQueues(queuesAsList); + }}); + this.pool.setQueues(queues); + } + @Test public void testGetJobFactory() { final Map> jobTypes = new LinkedHashMap>(); From e9fb8d7f44060aee2aa9d43a949beef2c4e1ed9b Mon Sep 17 00:00:00 2001 From: ofirnk Date: Tue, 29 Mar 2016 11:25:59 +0300 Subject: [PATCH 2/4] MultiplePriorityQueues: Fix comments --- .../net/greghaines/jesque/worker/Worker.java | 2 +- .../greghaines/jesque/worker/WorkerImpl.java | 78 +++++++------------ .../greghaines/jesque/worker/WorkerPool.java | 6 +- .../jesque/worker/TestWorkerPool.java | 14 ++-- 4 files changed, 39 insertions(+), 61 deletions(-) diff --git a/src/main/java/net/greghaines/jesque/worker/Worker.java b/src/main/java/net/greghaines/jesque/worker/Worker.java index a8f0413e..b84d32fc 100644 --- a/src/main/java/net/greghaines/jesque/worker/Worker.java +++ b/src/main/java/net/greghaines/jesque/worker/Worker.java @@ -99,7 +99,7 @@ public interface Worker extends JobExecutor, Runnable { * Clear any current queues and poll the given queues in the specified order. * @param queues the queues to poll */ - void setQueues(List queues); + void setOrderedPriorityQueues(List queues); /** * @return the worker event emitter for this worker diff --git a/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java b/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java index cdd55173..33fecf55 100644 --- a/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java +++ b/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java @@ -15,32 +15,18 @@ */ package net.greghaines.jesque.worker; -import com.fasterxml.jackson.core.JsonParseException; -import com.fasterxml.jackson.databind.JsonMappingException; -import net.greghaines.jesque.Config; -import net.greghaines.jesque.Job; -import net.greghaines.jesque.JobFailure; -import net.greghaines.jesque.WorkerStatus; -import net.greghaines.jesque.json.ObjectMapperFactory; -import net.greghaines.jesque.utils.JedisUtils; -import net.greghaines.jesque.utils.JesqueUtils; -import net.greghaines.jesque.utils.ScriptUtils; -import net.greghaines.jesque.utils.VersionUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import redis.clients.jedis.Jedis; -import redis.clients.jedis.exceptions.JedisException; +import static net.greghaines.jesque.utils.ResqueConstants.*; +import static net.greghaines.jesque.worker.JobExecutor.State.*; +import static net.greghaines.jesque.worker.WorkerEvent.*; +import static net.greghaines.jesque.worker.WorkerImpl.NextQueueStrategy.RESET_TO_HIGHEST_PRIORITY; import java.io.IOException; +import java.lang.Throwable; import java.lang.management.ManagementFactory; import java.net.InetAddress; import java.net.UnknownHostException; import java.text.SimpleDateFormat; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.Date; -import java.util.List; +import java.util.*; import java.util.concurrent.BlockingDeque; import java.util.concurrent.Callable; import java.util.concurrent.LinkedBlockingDeque; @@ -49,32 +35,24 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; -import static net.greghaines.jesque.utils.ResqueConstants.COLON; -import static net.greghaines.jesque.utils.ResqueConstants.DATE_FORMAT; -import static net.greghaines.jesque.utils.ResqueConstants.FAILED; -import static net.greghaines.jesque.utils.ResqueConstants.INFLIGHT; -import static net.greghaines.jesque.utils.ResqueConstants.JAVA_DYNAMIC_QUEUES; -import static net.greghaines.jesque.utils.ResqueConstants.PROCESSED; -import static net.greghaines.jesque.utils.ResqueConstants.QUEUE; -import static net.greghaines.jesque.utils.ResqueConstants.QUEUES; -import static net.greghaines.jesque.utils.ResqueConstants.STARTED; -import static net.greghaines.jesque.utils.ResqueConstants.STAT; -import static net.greghaines.jesque.utils.ResqueConstants.WORKER; -import static net.greghaines.jesque.utils.ResqueConstants.WORKERS; -import static net.greghaines.jesque.worker.JobExecutor.State.NEW; -import static net.greghaines.jesque.worker.JobExecutor.State.RUNNING; -import static net.greghaines.jesque.worker.JobExecutor.State.SHUTDOWN; -import static net.greghaines.jesque.worker.JobExecutor.State.SHUTDOWN_IMMEDIATE; -import static net.greghaines.jesque.worker.WorkerEvent.JOB_EXECUTE; -import static net.greghaines.jesque.worker.WorkerEvent.JOB_FAILURE; -import static net.greghaines.jesque.worker.WorkerEvent.JOB_PROCESS; -import static net.greghaines.jesque.worker.WorkerEvent.JOB_SUCCESS; -import static net.greghaines.jesque.worker.WorkerEvent.WORKER_ERROR; -import static net.greghaines.jesque.worker.WorkerEvent.WORKER_POLL; -import static net.greghaines.jesque.worker.WorkerEvent.WORKER_START; -import static net.greghaines.jesque.worker.WorkerEvent.WORKER_STOP; -import static net.greghaines.jesque.worker.WorkerImpl.NextQueueStrategy.DRAIN_WHILE_MESSAGES_EXISTS; -import static net.greghaines.jesque.worker.WorkerImpl.NextQueueStrategy.RESET_TO_HIGHEST_PRIORITY; +import net.greghaines.jesque.Config; +import net.greghaines.jesque.Job; +import net.greghaines.jesque.JobFailure; +import net.greghaines.jesque.WorkerStatus; +import net.greghaines.jesque.json.ObjectMapperFactory; +import net.greghaines.jesque.utils.JedisUtils; +import net.greghaines.jesque.utils.JesqueUtils; +import net.greghaines.jesque.utils.ScriptUtils; +import net.greghaines.jesque.utils.VersionUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import redis.clients.jedis.Jedis; +import redis.clients.jedis.exceptions.JedisException; + +import com.fasterxml.jackson.core.JsonParseException; +import com.fasterxml.jackson.databind.JsonMappingException; /** * WorkerImpl is an implementation of the Worker interface. Obeys the contract of a Resque worker in Redis. @@ -173,7 +151,7 @@ public WorkerImpl(final Config config, final Collection queues, final Jo (queues == null ? Collections.EMPTY_LIST : new ArrayList<>(queues)), jobFactory, jedis, - DRAIN_WHILE_MESSAGES_EXISTS); + NextQueueStrategy.DRAIN_WHILE_MESSAGES_EXISTS); } /** @@ -206,7 +184,7 @@ public WorkerImpl(final Config config, final List queues, final JobFacto this.failQueueStrategyRef = new AtomicReference( new DefaultFailQueueStrategy(this.namespace)); authenticateAndSelectDB(); - setQueues(queues); + setOrderedPriorityQueues(queues); this.name = createName(); } @@ -373,14 +351,14 @@ public void removeAllQueues() { */ @Override public void setQueues(final Collection queues) { - setQueues(new ArrayList<>(queues)); + setOrderedPriorityQueues(new ArrayList<>(queues)); } /** * {@inheritDoc} */ @Override - public void setQueues(final List queues) { + public void setOrderedPriorityQueues(final List queues) { checkQueues(queues); this.queueNames.clear(); this.queueNames.addAll((queues == ALL_QUEUES) // Using object equality on purpose diff --git a/src/main/java/net/greghaines/jesque/worker/WorkerPool.java b/src/main/java/net/greghaines/jesque/worker/WorkerPool.java index f9fe327c..e52fc91b 100644 --- a/src/main/java/net/greghaines/jesque/worker/WorkerPool.java +++ b/src/main/java/net/greghaines/jesque/worker/WorkerPool.java @@ -253,16 +253,16 @@ public void removeAllQueues() { */ @Override public void setQueues(final Collection queues) { - setQueues(new ArrayList<>(queues)); + setOrderedPriorityQueues(new ArrayList<>(queues)); } /** * {@inheritDoc} */ @Override - public void setQueues(final List queues) { + public void setOrderedPriorityQueues(final List queues) { for (final Worker worker : this.workers) { - worker.setQueues(queues); + worker.setOrderedPriorityQueues(queues); } } diff --git a/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java b/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java index bbce8004..47ad7c04 100644 --- a/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java +++ b/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java @@ -214,24 +214,24 @@ public void testRemoveAllQueues() { } @Test - public void testSetQueues() { + public void testSetQueuesList() { final List queues = Arrays.asList("queue1", "queue2"); this.mockCtx.checking(new Expectations(){{ - oneOf(workers.get(0)).setQueues(queues); - oneOf(workers.get(1)).setQueues(queues); + oneOf(workers.get(0)).setOrderedPriorityQueues(queues); + oneOf(workers.get(1)).setOrderedPriorityQueues(queues); }}); - this.pool.setQueues(queues); + this.pool.setOrderedPriorityQueues(queues); } @Test - public void testSetCollectionQueues() { + public void testSetQueuesCollection() { final Collection queues = new HashSet<>(); queues.add("queue1"); queues.add("queue2"); final List queuesAsList = Arrays.asList("queue1", "queue2"); this.mockCtx.checking(new Expectations(){{ - oneOf(workers.get(0)).setQueues(queuesAsList); - oneOf(workers.get(1)).setQueues(queuesAsList); + oneOf(workers.get(0)).setOrderedPriorityQueues(queuesAsList); + oneOf(workers.get(1)).setOrderedPriorityQueues(queuesAsList); }}); this.pool.setQueues(queues); } From f12f23d409a4d402032f0d1ed1998adce80eba45 Mon Sep 17 00:00:00 2001 From: ofirnk Date: Mon, 31 Oct 2016 19:00:44 +0200 Subject: [PATCH 3/4] Syncing with kenshoo/jesque/pull/6 --- .../greghaines/jesque/worker/WorkerImpl.java | 11 ++++-- .../fromMultiplePriorityQueues.lua | 34 +++++++++---------- .../jesque/worker/TestWorkerImpl.java | 14 ++++---- .../jesque/worker/TestWorkerPool.java | 18 +++++----- 4 files changed, 41 insertions(+), 36 deletions(-) diff --git a/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java b/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java index 33fecf55..d088c027 100644 --- a/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java +++ b/src/main/java/net/greghaines/jesque/worker/WorkerImpl.java @@ -15,18 +15,22 @@ */ package net.greghaines.jesque.worker; + import static net.greghaines.jesque.utils.ResqueConstants.*; import static net.greghaines.jesque.worker.JobExecutor.State.*; import static net.greghaines.jesque.worker.WorkerEvent.*; import static net.greghaines.jesque.worker.WorkerImpl.NextQueueStrategy.RESET_TO_HIGHEST_PRIORITY; import java.io.IOException; -import java.lang.Throwable; import java.lang.management.ManagementFactory; import java.net.InetAddress; import java.net.UnknownHostException; import java.text.SimpleDateFormat; -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Date; +import java.util.List; import java.util.concurrent.BlockingDeque; import java.util.concurrent.Callable; import java.util.concurrent.LinkedBlockingDeque; @@ -66,6 +70,7 @@ public class WorkerImpl implements Worker { protected static final int RECONNECT_ATTEMPTS = 120; // Total time: 10 min private static final String LPOPLPUSH_LUA = "/workerScripts/jesque_lpoplpush.lua"; private static final String POP_LUA = "/workerScripts/jesque_pop.lua"; + private static final String POP_FROM_MULTIPLE_PRIO_QUEUES = "/workerScripts/fromMultiplePriorityQueues.lua"; // Set the thread name to the message for debugging private static volatile boolean threadNameChangingEnabled = false; @@ -210,7 +215,7 @@ public void run() { this.listenerDelegate.fireEvent(WORKER_START, this, null, null, null, null, null); this.popScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(POP_LUA))); this.lpoplpushScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(LPOPLPUSH_LUA))); - this.multiPriorityQueuesScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(LPOPLPUSH_LUA))); + this.multiPriorityQueuesScriptHash.set(this.jedis.scriptLoad(ScriptUtils.readScript(POP_FROM_MULTIPLE_PRIO_QUEUES))); poll(); } catch (Exception ex) { LOG.error("Uncaught exception in worker run-loop!", ex); diff --git a/src/main/resources/workerScripts/fromMultiplePriorityQueues.lua b/src/main/resources/workerScripts/fromMultiplePriorityQueues.lua index 1cbc15b0..4a06f6a0 100644 --- a/src/main/resources/workerScripts/fromMultiplePriorityQueues.lua +++ b/src/main/resources/workerScripts/fromMultiplePriorityQueues.lua @@ -1,4 +1,4 @@ --- +-- new -- User: Ofir Naor -- Date: 3/27/16 -- Time: 2:47 PM @@ -12,23 +12,23 @@ local OPTIONAL_SPACE_SEPARATOR = '%s*' local NEXT_QUEUE_REGEX = QUEUE_NAME_CAPTURING_REGEX .. OPTIONAL_COMMA_SEPARATOR .. OPTIONAL_SPACE_SEPARATOR for q in queues.gmatch(queues, NEXT_QUEUE_REGEX) do - local firstMsg = redis.call('ZRANGE', q, '0', '0') --- table.insert(debug, "q: " .. q) - if firstMsg ~= nil then --- table.insert(debug, 'result:') --- table.insert(debug, firstMsg) --- table.insert(debug, 'removed:') - local payload = firstMsg[1] - if payload ~= nil then - local removedItems = redis.call('ZREM', q, payload) --- table.insert(debug, payload) --- table.insert(debug, "num of removed items:") --- table.insert(debug, type(payload)) - return payload --- return debug + local queueName = 'resque:queue:' .. q + local status, queueType = next(redis.call('TYPE', queueName)) + local payload + if queueType == 'zset' then + local firstMsg = redis.call('ZRANGE', queueName, '0', '0') + if firstMsg ~= nil then + payload = firstMsg[1] + if payload ~= nil then + local removedItems = redis.call('ZREM', queueName, payload) + end end + elseif queueType == 'list' then + payload = redis.call('LPOP', queueName) + end + + if payload ~= nil then + return payload end end ---return debug ---return "NOT_FOUND" return nil \ No newline at end of file diff --git a/src/test/java/net/greghaines/jesque/worker/TestWorkerImpl.java b/src/test/java/net/greghaines/jesque/worker/TestWorkerImpl.java index f5f745cd..2fed28b8 100644 --- a/src/test/java/net/greghaines/jesque/worker/TestWorkerImpl.java +++ b/src/test/java/net/greghaines/jesque/worker/TestWorkerImpl.java @@ -1,5 +1,12 @@ package net.greghaines.jesque.worker; +import static net.greghaines.jesque.utils.JesqueUtils.entry; +import static net.greghaines.jesque.utils.JesqueUtils.map; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; + import net.greghaines.jesque.Config; import net.greghaines.jesque.ConfigBuilder; import net.greghaines.jesque.TestAction; @@ -11,13 +18,6 @@ import org.junit.Test; import redis.clients.jedis.Jedis; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; - -import static net.greghaines.jesque.utils.JesqueUtils.entry; -import static net.greghaines.jesque.utils.JesqueUtils.map; - public class TestWorkerImpl { private static final Config CONFIG = new ConfigBuilder().build(); diff --git a/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java b/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java index 47ad7c04..13c2c944 100644 --- a/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java +++ b/src/test/java/net/greghaines/jesque/worker/TestWorkerPool.java @@ -1,14 +1,5 @@ package net.greghaines.jesque.worker; -import org.jmock.Expectations; -import org.jmock.Mockery; -import org.jmock.integration.junit4.JUnit4Mockery; -import org.jmock.lib.concurrent.Synchroniser; -import org.junit.After; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; - import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -18,6 +9,15 @@ import java.util.Map; import java.util.concurrent.Callable; +import org.jmock.Expectations; +import org.jmock.Mockery; +import org.jmock.integration.junit4.JUnit4Mockery; +import org.jmock.lib.concurrent.Synchroniser; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + public class TestWorkerPool { private static final int NUM_WORKERS = 2; From f55e59b642fa1166043edf99959eee6eeb59d271 Mon Sep 17 00:00:00 2001 From: ofirnk Date: Tue, 1 Nov 2016 15:33:02 +0200 Subject: [PATCH 4/4] Make Jesque compile and throw unsupported operation exception for trying to use WorkerPoolimpl with MultiplePriorityQueues --- .../net/greghaines/jesque/worker/WorkerPoolImpl.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java b/src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java index 5754d19f..3f40fa78 100644 --- a/src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java +++ b/src/main/java/net/greghaines/jesque/worker/WorkerPoolImpl.java @@ -25,10 +25,7 @@ import java.net.InetAddress; import java.net.UnknownHostException; import java.text.SimpleDateFormat; -import java.util.Collection; -import java.util.Collections; -import java.util.Date; -import java.util.Set; +import java.util.*; import java.util.concurrent.BlockingDeque; import java.util.concurrent.Callable; import java.util.concurrent.LinkedBlockingDeque; @@ -352,6 +349,11 @@ public Set doWork(final Jedis jedis) { } } + @Override + public void setOrderedPriorityQueues(List queues) { + throw new UnsupportedOperationException("Not implemented, yet"); + } + /** * {@inheritDoc} */