diff --git a/app/models/ci/job_artifact.rb b/app/models/ci/job_artifact.rb
index 11c88200c37..789bb293811 100644
--- a/app/models/ci/job_artifact.rb
+++ b/app/models/ci/job_artifact.rb
@@ -73,6 +73,8 @@ module Ci
       where(file_type: types)
     end
 
+    scope :expired, -> (limit) { where('expire_at < ?', Time.now).limit(limit) }
+
     delegate :filename, :exists?, :open, to: :file
 
     enum file_type: {
diff --git a/app/services/ci/destroy_expired_job_artifacts_service.rb b/app/services/ci/destroy_expired_job_artifacts_service.rb
new file mode 100644
index 00000000000..7d2f5d33fed
--- /dev/null
+++ b/app/services/ci/destroy_expired_job_artifacts_service.rb
@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+module Ci
+  class DestroyExpiredJobArtifactsService
+    include ::Gitlab::ExclusiveLeaseHelpers
+    include ::Gitlab::LoopHelpers
+
+    BATCH_SIZE = 100
+    LOOP_TIMEOUT = 45.minutes
+    LOOP_LIMIT = 1000
+    EXCLUSIVE_LOCK_KEY = 'expired_job_artifacts:destroy:lock'
+    LOCK_TIMEOUT = 50.minutes
+
+    ##
+    # Destroy expired job artifacts on GitLab instance
+    #
+    # This destroy process cannot run for more than 45 minutes. This is for
+    # preventing multiple `ExpireBuildArtifactsWorker` CRON jobs run concurrently,
+    # which is scheduled at every hour.
+    def execute
+      in_lock(EXCLUSIVE_LOCK_KEY, ttl: LOCK_TIMEOUT, retries: 1) do
+        loop_until(timeout: LOOP_TIMEOUT, limit: LOOP_LIMIT) do
+          destroy_batch
+        end
+      end
+    end
+
+    private
+
+    def destroy_batch
+      artifacts = Ci::JobArtifact.expired(BATCH_SIZE).to_a
+
+      return false if artifacts.empty?
+
+      artifacts.each(&:destroy!)
+    end
+  end
+end
diff --git a/app/workers/expire_build_artifacts_worker.rb b/app/workers/expire_build_artifacts_worker.rb
index dce812d1ae2..251e95c68d5 100644
--- a/app/workers/expire_build_artifacts_worker.rb
+++ b/app/workers/expire_build_artifacts_worker.rb
@@ -4,8 +4,20 @@ class ExpireBuildArtifactsWorker
   include ApplicationWorker
   include CronjobQueue
 
-  # rubocop: disable CodeReuse/ActiveRecord
   def perform
+    if Feature.enabled?(:ci_new_expire_job_artifacts_service, default_enabled: true)
+      perform_efficient_artifacts_removal
+    else
+      perform_legacy_artifacts_removal
+    end
+  end
+
+  def perform_efficient_artifacts_removal
+    Ci::DestroyExpiredJobArtifactsService.new.execute
+  end
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def perform_legacy_artifacts_removal
     Rails.logger.info 'Scheduling removal of build artifacts'
 
     build_ids = Ci::Build.with_expired_artifacts.pluck(:id)
diff --git a/changelogs/unreleased/expire-job-artifacts-worker.yml b/changelogs/unreleased/expire-job-artifacts-worker.yml
new file mode 100644
index 00000000000..cda6e9ff497
--- /dev/null
+++ b/changelogs/unreleased/expire-job-artifacts-worker.yml
@@ -0,0 +1,5 @@
+---
+title: Efficiently remove expired artifacts in `ExpireBuildArtifactsWorker`
+merge_request: 24450
+author:
+type: performance
diff --git a/lib/gitlab/loop_helpers.rb b/lib/gitlab/loop_helpers.rb
new file mode 100644
index 00000000000..3873156a3b0
--- /dev/null
+++ b/lib/gitlab/loop_helpers.rb
@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module LoopHelpers
+    ##
+    # This helper method repeats the same task until it's expired.
+    #
+    # Note: ExpiredLoopError does not happen until the given block finished.
+    #       Please do not use this method for heavy or asynchronous operations.
+    def loop_until(timeout: nil, limit: 1_000_000)
+      raise ArgumentError unless limit
+
+      start = Time.now
+
+      limit.times do
+        return true unless yield
+
+        return false if timeout && (Time.now - start) > timeout
+      end
+
+      false
+    end
+  end
+end
diff --git a/spec/lib/gitlab/loop_helpers_spec.rb b/spec/lib/gitlab/loop_helpers_spec.rb
new file mode 100644
index 00000000000..e17a0342d64
--- /dev/null
+++ b/spec/lib/gitlab/loop_helpers_spec.rb
@@ -0,0 +1,45 @@
+require 'spec_helper'
+
+describe Gitlab::LoopHelpers do
+  let(:class_instance) { (Class.new { include ::Gitlab::LoopHelpers }).new }
+
+  describe '#loop_until' do
+    subject do
+      class_instance.loop_until(**params) { true }
+    end
+
+    context 'when limit is not given' do
+      let(:params) { { limit: nil } }
+
+      it 'raises an error' do
+        expect { subject }.to raise_error(ArgumentError)
+      end
+    end
+
+    context 'when timeout is specified' do
+      let(:params) { { timeout: 1.second } }
+
+      it "returns false after it's expired" do
+        is_expected.to be_falsy
+      end
+
+      it 'executes the block at least once' do
+        expect { |b| class_instance.loop_until(**params, &b) }
+          .to yield_control.at_least(1)
+      end
+    end
+
+    context 'when iteration limit is specified' do
+      let(:params) { { limit: 1 } }
+
+      it "returns false after it's expired" do
+        is_expected.to be_falsy
+      end
+
+      it 'executes the block once' do
+        expect { |b| class_instance.loop_until(**params, &b) }
+          .to yield_control.once
+      end
+    end
+  end
+end
diff --git a/spec/services/ci/destroy_expired_job_artifacts_service_spec.rb b/spec/services/ci/destroy_expired_job_artifacts_service_spec.rb
new file mode 100644
index 00000000000..80d82ba3ac9
--- /dev/null
+++ b/spec/services/ci/destroy_expired_job_artifacts_service_spec.rb
@@ -0,0 +1,104 @@
+require 'spec_helper'
+
+describe Ci::DestroyExpiredJobArtifactsService, :clean_gitlab_redis_shared_state do
+  include ExclusiveLeaseHelpers
+
+  describe '.execute' do
+    subject { service.execute }
+
+    let(:service) { described_class.new }
+    let!(:artifact) { create(:ci_job_artifact, expire_at: 1.day.ago) }
+
+    it 'destroys expired job artifacts' do
+      expect { subject }.to change { Ci::JobArtifact.count }.by(-1)
+    end
+
+    context 'when artifact is not expired' do
+      let!(:artifact) { create(:ci_job_artifact, expire_at: 1.day.since) }
+
+      it 'does not destroy expired job artifacts' do
+        expect { subject }.not_to change { Ci::JobArtifact.count }
+      end
+    end
+
+    context 'when artifact is permanent' do
+      let!(:artifact) { create(:ci_job_artifact, expire_at: nil) }
+
+      it 'does not destroy expired job artifacts' do
+        expect { subject }.not_to change { Ci::JobArtifact.count }
+      end
+    end
+
+    context 'when failed to destroy artifact' do
+      before do
+        stub_const('Ci::DestroyExpiredJobArtifactsService::LOOP_LIMIT', 10)
+
+        allow_any_instance_of(Ci::JobArtifact)
+          .to receive(:destroy!)
+          .and_raise(ActiveRecord::RecordNotDestroyed)
+      end
+
+      it 'raises an exception and stop destroying' do
+        expect { subject }.to raise_error(ActiveRecord::RecordNotDestroyed)
+      end
+    end
+
+    context 'when exclusive lease has already been taken by the other instance' do
+      before do
+        stub_exclusive_lease_taken(described_class::EXCLUSIVE_LOCK_KEY, timeout: described_class::LOCK_TIMEOUT)
+      end
+
+      it 'raises an error and does not start destroying' do
+        expect { subject }.to raise_error(Gitlab::ExclusiveLeaseHelpers::FailedToObtainLockError)
+      end
+    end
+
+    context 'when timeout happens' do
+      before do
+        stub_const('Ci::DestroyExpiredJobArtifactsService::LOOP_TIMEOUT', 1.second)
+        allow_any_instance_of(described_class).to receive(:destroy_batch) { true }
+      end
+
+      it 'returns false and does not continue destroying' do
+        is_expected.to be_falsy
+      end
+    end
+
+    context 'when loop reached loop limit' do
+      before do
+        stub_const('Ci::DestroyExpiredJobArtifactsService::LOOP_LIMIT', 1)
+        stub_const('Ci::DestroyExpiredJobArtifactsService::BATCH_SIZE', 1)
+      end
+
+      let!(:artifact) { create_list(:ci_job_artifact, 2, expire_at: 1.day.ago) }
+
+      it 'raises an error and does not continue destroying' do
+        is_expected.to be_falsy
+      end
+
+      it 'destroys one artifact' do
+        expect { subject }.to change { Ci::JobArtifact.count }.by(-1)
+      end
+    end
+
+    context 'when there are no artifacts' do
+      let!(:artifact) { }
+
+      it 'does not raise error' do
+        expect { subject }.not_to raise_error
+      end
+    end
+
+    context 'when there are artifacts more than batch sizes' do
+      before do
+        stub_const('Ci::DestroyExpiredJobArtifactsService::BATCH_SIZE', 1)
+      end
+
+      let!(:artifact) { create_list(:ci_job_artifact, 2, expire_at: 1.day.ago) }
+
+      it 'destroys all expired artifacts' do
+        expect { subject }.to change { Ci::JobArtifact.count }.by(-2)
+      end
+    end
+  end
+end
diff --git a/spec/workers/expire_build_artifacts_worker_spec.rb b/spec/workers/expire_build_artifacts_worker_spec.rb
index b47b4a02a68..27995cf1611 100644
--- a/spec/workers/expire_build_artifacts_worker_spec.rb
+++ b/spec/workers/expire_build_artifacts_worker_spec.rb
@@ -11,6 +11,7 @@ describe ExpireBuildArtifactsWorker do
 
   describe '#perform' do
     before do
+      stub_feature_flags(ci_new_expire_job_artifacts_service: false)
       build
     end
 
@@ -47,4 +48,17 @@ describe ExpireBuildArtifactsWorker do
       Sidekiq::Queues.jobs_by_worker['ExpireBuildInstanceArtifactsWorker']
     end
   end
+
+  describe '#perform with ci_new_expire_job_artifacts_service feature flag' do
+    before do
+      stub_feature_flags(ci_new_expire_job_artifacts_service: true)
+    end
+
+    it 'executes a service' do
+      expect_any_instance_of(Ci::DestroyExpiredJobArtifactsService).to receive(:execute)
+      expect(ExpireBuildInstanceArtifactsWorker).not_to receive(:bulk_perform_async)
+
+      worker.perform
+    end
+  end
 end