diff --git a/lib/delayed/backend/job_preparer.rb b/lib/delayed/backend/job_preparer.rb
index 24c869752..5505c9fab 100644
--- a/lib/delayed/backend/job_preparer.rb
+++ b/lib/delayed/backend/job_preparer.rb
@@ -33,8 +33,12 @@ def set_queue_name
       end
 
       def set_priority
-        queue_attribute = Delayed::Worker.queue_attributes[options[:queue]]
-        options[:priority] ||= (queue_attribute && queue_attribute[:priority]) || Delayed::Worker.default_priority
+        options[:priority] ||= if options[:payload_object].respond_to?(:priority)
+          options[:payload_object].priority
+        else
+          queue_attribute = Delayed::Worker.queue_attributes[options[:queue]]
+          (queue_attribute && queue_attribute[:priority]) || Delayed::Worker.default_priority
+        end
       end
 
       def handle_deprecation
diff --git a/lib/delayed/backend/shared_spec.rb b/lib/delayed/backend/shared_spec.rb
index 211187265..eb2cf9e1d 100644
--- a/lib/delayed/backend/shared_spec.rb
+++ b/lib/delayed/backend/shared_spec.rb
@@ -76,6 +76,11 @@ def create_job(opts = {})
         job = described_class.enqueue :payload_object => NamedQueueJob.new
         expect(job.queue).to eq(NamedQueueJob.new.queue_name)
       end
+
+      it "uses the payload object's priority" do
+        job = described_class.enqueue :payload_object => PrioritizedQueueJob.new
+        expect(job.priority).to eq(PrioritizedQueueJob.new.priority)
+      end
     end
 
     context 'with multiple arguments' do
@@ -333,6 +338,23 @@ def create_job(opts = {})
       job = described_class.enqueue :payload_object => NamedQueueJob.new, :priority => 10
       expect(job.priority).to eq(10)
     end
+
+    it 'sets job priority based on queue_attributes configuration' do
+      Delayed::Worker.queue_attributes = {'job_tracking' => {:priority => 4}}
+      job = described_class.enqueue :payload_object => PrioritizedQueueJob.new
+      expect(job.priority).to eq(5)
+    end
+
+    it 'sets job priority based on the passed in priority overrideing queue_attributes configuration' do
+      Delayed::Worker.queue_attributes = {'job_tracking' => {:priority => 4}}
+      job = described_class.enqueue :payload_object => PrioritizedQueueJob.new, :priority => 10
+      expect(job.priority).to eq(10)
+    end
+
+    it 'sets job priority based on the passed in priority in job class' do
+      job = described_class.enqueue :payload_object => PrioritizedQueueJob.new
+      expect(job.priority).to eq(5)
+    end
   end
 
   context 'clear_locks!' do
diff --git a/spec/sample_jobs.rb b/spec/sample_jobs.rb
index 36ba136ab..43eda86dc 100644
--- a/spec/sample_jobs.rb
+++ b/spec/sample_jobs.rb
@@ -19,6 +19,12 @@ def queue_name
   end
 end
 
+class PrioritizedQueueJob < SimpleJob
+  def priority
+    5
+  end
+end
+
 class ErrorJob
   cattr_accessor :runs
   @runs = 0