diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb
index 03605c24..a3860e1f 100644
--- a/app/models/solid_queue/job/executable.rb
+++ b/app/models/solid_queue/job/executable.rb
@@ -39,8 +39,12 @@ def dispatch_all_at_once(jobs)
             ReadyExecution.create_all_from_jobs jobs
           end
 
+          # Each job acquires its concurrency lock here, and holds it until the
+          # transaction dispatching the whole set ends. Going through them by
+          # concurrency key makes everyone dispatching at the same time acquire
+          # the locks they share in the same order, so they can't deadlock.
           def dispatch_all_one_by_one(jobs)
-            jobs.each(&:dispatch)
+            jobs.sort_by { |job| [ job.concurrency_key, job.id ] }.each(&:dispatch)
           end
 
           def successfully_dispatched(jobs)
diff --git a/test/models/solid_queue/job_test.rb b/test/models/solid_queue/job_test.rb
index 2042a30d..cfd604f6 100644
--- a/test/models/solid_queue/job_test.rb
+++ b/test/models/solid_queue/job_test.rb
@@ -245,6 +245,21 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob
     assert_not not_enqueued.successfully_enqueued?
   end
 
+  test "enqueue jobs in bulk acquiring their concurrency locks by key" do
+    results = 3.times.map { JobResult.create!(queue_name: "default") }
+
+    acquired = []
+    SolidQueue::Semaphore.stubs(:wait).with { |job| acquired << [ job.concurrency_key, job.id ] }.returns(true)
+
+    # Two jobs for each key, with the keys in descending order
+    active_jobs = (results.reverse + results.reverse).map { |result| NonOverlappingJob.new(result) }
+    ActiveJob.perform_all_later(active_jobs)
+
+    # By key and, within a key, in the order they were enqueued
+    assert_equal 6, acquired.size
+    assert_equal acquired.sort, acquired
+  end
+
   test "discard ready job" do
     AddToBufferJob.perform_later(1)
     job = SolidQueue::Job.last
diff --git a/test/models/solid_queue/semaphore_test.rb b/test/models/solid_queue/semaphore_test.rb
index 432b97af..2877afd2 100644
--- a/test/models/solid_queue/semaphore_test.rb
+++ b/test/models/solid_queue/semaphore_test.rb
@@ -5,6 +5,13 @@
 class SolidQueue::SemaphoreTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
+  class NonOverlappingByKeyJob < ApplicationJob
+    limits_concurrency key: ->(key) { key }
+
+    def perform(key)
+    end
+  end
+
   setup do
     @result = JobResult.create!(queue_name: "default")
   end
@@ -108,7 +115,73 @@ class SolidQueue::SemaphoreTest < ActiveSupport::TestCase
     assert_equal 0, SolidQueue::BlockedExecution.where(concurrency_key: concurrency_key).count
   end
 
+  test "dispatchers with jobs for the same concurrency keys in different order don't deadlock" do
+    skip_on_sqlite
+
+    # By job id: one, two, two, one. In batches of two, a dispatcher gets the
+    # first two jobs and another one, skipping those, the other two.
+    jobs = %w[ one two two one ].map do |key|
+      NonOverlappingByKeyJob.set(wait: 1.minute).perform_later(key)
+      SolidQueue::Job.last
+    end
+    travel_to 2.minutes.from_now
+
+    errors = dispatch_in_two_batches_at_once
+
+    assert_empty errors
+    assert_equal 0, SolidQueue::ScheduledExecution.count
+
+    # The first job for each key gets the lock
+    assert_equal [ :ready, :ready, :blocked, :blocked ], jobs.map { |job| job.reload.status }
+  end
+
   private
+    # Two dispatchers, each with a batch of two jobs. Each one, after acquiring
+    # its first lock, gives the other the chance to acquire its first one too,
+    # so that both hold a lock when they go for the second.
+    def dispatch_in_two_batches_at_once
+      acquired_first = [ Concurrent::Event.new, Concurrent::Event.new ]
+
+      pause_after_first_lock(acquired_first) do
+        dispatchers = 2.times.map do |index|
+          Thread.new do
+            Thread.current[:dispatcher] = index
+            # The second one starts once the first has its batch and its first lock
+            acquired_first[0].wait(5.seconds) if index == 1
+
+            SolidQueue::Record.connection_pool.with_connection do
+              SolidQueue::ScheduledExecution.dispatch_next_batch(2)
+            end
+
+            nil
+          rescue => error
+            error
+          end
+        end
+
+        dispatchers.map(&:value).compact
+      end
+    end
+
+    def pause_after_first_lock(acquired_first)
+      original_wait = SolidQueue::Semaphore.method(:wait)
+
+      SolidQueue::Semaphore.define_singleton_method(:wait) do |job|
+        original_wait.call(job).tap do
+          index = Thread.current[:dispatcher]
+
+          if index && !acquired_first[index].set?
+            acquired_first[index].set
+            acquired_first[1 - index].wait(2.seconds)
+          end
+        end
+      end
+
+      yield
+    ensure
+      SolidQueue::Semaphore.define_singleton_method(:wait, original_wait)
+    end
+
     def skip_on_sqlite
       skip "Row-level locking not supported on SQLite" if SolidQueue::Record.connection.adapter_name.downcase.include?("sqlite")
     end