diff --git a/README.md b/README.md
index d57320d..4617fce 100644
--- a/README.md
+++ b/README.md
@@ -228,6 +228,11 @@ These let you tweak the default behavior.
 Most often, you'll just be setting the `cache_duration`
 to enable result caching.
 
+#### always_start_new_herd
+
+Skip herd avoidance check if True, but still trigger avoidance for other tasks.
+Defaults to False.
+
 #### cache_duration
 
 If you want your results cached,
diff --git a/jobtastic/task.py b/jobtastic/task.py
index edcb814..947d41e 100644
--- a/jobtastic/task.py
+++ b/jobtastic/task.py
@@ -82,6 +82,7 @@ class JobtasticTask(Task):
       front-end code so that users know what to expect.
     """
     abstract = True
+    always_start_new_herd = False
 
     #: The shared cache used for locking and thundering herd protection
     _cache = None
@@ -218,11 +219,13 @@ def apply_async(self, args, kwargs, **options):
                 'Found existing cached and completed task: %s', task_id)
             return self.AsyncResult(task_id)
 
-        # Check for an in-progress equivalent task to avoid duplicating work
-        task_id = self.cache.get('herd:%s' % cache_key)
-        if task_id:
-            logging.info('Found existing in-progress task: %s', task_id)
-            return self.AsyncResult(task_id)
+        if not self.always_start_new_herd:
+            # Check for an in-progress equivalent task to avoid
+            # duplicating work
+            task_id = self.cache.get('herd:%s' % cache_key)
+            if task_id:
+                logging.info('Found existing in-progress task: %s', task_id)
+                return self.AsyncResult(task_id)
 
         # It's not cached and it's not already running. Use an atomic lock to
         # start the task, ensuring there isn't a race condition that could