diff --git a/CHANGELOG.md b/CHANGELOG.md
index 35125fb7..35f2a686 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,12 @@ This file will no longer be updated - all changes after v6.7.0 will only be docu
 - Support rack 3 by @ioquatix in #586
 - Gem release management. by @ioquatix in #614
 
+## [6.x.x] = 2022-xx-xx
+
+### Added
+
+- Added pseudo-random time offsets to throttling. If your application uses a custom throttle lambda to emit RateLimit-style headers, see the README for updated sample code.
+
 ## [6.6.1] - 2022-04-14
 
 ### Fixed
diff --git a/README.md b/README.md
index 545003cc..aa7fdf1c 100644
--- a/README.md
+++ b/README.md
@@ -337,6 +337,7 @@ Rack::Attack.throttled_responder = lambda do |request|
   # DOSed the site. Rack::Attack returns 429 for throttling by default
   [ 503, {}, ["Server Error\n"]]
 end
+Rack::Attack.configuration.throttled_responder_is_offset_aware = true
 ```
 
 ### RateLimit headers for well-behaved clients
@@ -344,12 +345,12 @@ end
 While Rack::Attack's primary focus is minimizing harm from abusive clients, it
 can also be used to return rate limit data that's helpful for well-behaved clients.
 
-If you want to return to user how many seconds to wait until they can start sending requests again, this can be done through enabling `Retry-After` header:
+If you want to report to the client how many seconds to wait until they can start sending requests again, per RFCs 6585 and 7231, this can be done through enabling the `Retry-After` header:
 ```ruby
 Rack::Attack.throttled_response_retry_after_header = true
 ```
 
-Here's an example response that includes conventional `RateLimit-*` headers:
+If you prefer to emit one of the RateLimit-style standards, you might write your own lambda like this (this example uses the [IETF WG standard](https://github.com/ietf-wg-httpapi/ratelimit-headers)):
 
 ```ruby
 Rack::Attack.throttled_responder = lambda do |request|
@@ -359,18 +360,18 @@ Rack::Attack.throttled_responder = lambda do |request|
   headers = {
     'RateLimit-Limit' => match_data[:limit].to_s,
     'RateLimit-Remaining' => '0',
-    'RateLimit-Reset' => (now + (match_data[:period] - now % match_data[:period])).to_s
+    'RateLimit-Reset' => (match_data[:retry_after] - now).to_s
   }
 
   [ 429, headers, ["Throttled\n"]]
 end
+Rack::Attack.configuration.throttled_responder_is_offset_aware = true
 ```
 
-
-For responses that did not exceed a throttle limit, Rack::Attack annotates the env with match data:
+For responses that exceeded a throttle limit, Rack::Attack annotates the env with match data:
 
 ```ruby
-request.env['rack.attack.throttle_data'][name] # => { discriminator: d, count: n, period: p, limit: l, epoch_time: t }
+request.env['rack.attack.throttle_data'][name] # => { discriminator: d, count: n, period: p, limit: l, epoch_time: t, retry_after: r }
 ```
 
 ## Logging & Instrumentation
diff --git a/lib/rack/attack/cache.rb b/lib/rack/attack/cache.rb
index ecbd3368..ce8c10d1 100644
--- a/lib/rack/attack/cache.rb
+++ b/lib/rack/attack/cache.rb
@@ -1,10 +1,13 @@
 # frozen_string_literal: true
 
+require 'digest'
+
 module Rack
   class Attack
     class Cache
       attr_accessor :prefix
       attr_reader :last_epoch_time
+      attr_reader :last_retry_after_time
 
       def self.default_store
         if Object.const_defined?(:Rails) && Rails.respond_to?(:cache)
@@ -28,8 +31,8 @@ def store=(store)
           end
       end
 
-      def count(unprefixed_key, period)
-        key, expires_in = key_and_expiry(unprefixed_key, period)
+      def count(unprefixed_key, period, use_offset = false)
+        key, expires_in = key_and_expiry(unprefixed_key, period, use_offset)
         do_count(key, expires_in)
       end
 
@@ -66,11 +69,23 @@ def reset!
 
       private
 
-      def key_and_expiry(unprefixed_key, period)
+      def key_and_expiry(unprefixed_key, period, use_offset = false)
         @last_epoch_time = Time.now.to_i
+        offset = offset_for(unprefixed_key, period, use_offset)
+        period_number, time_into_period = period_number_and_time_into(period, offset)
+        period_remainder = period - time_into_period
+        @last_retry_after_time = @last_epoch_time + period_remainder
         # Add 1 to expires_in to avoid timing error: https://github.com/rack/rack-attack/pull/85
-        expires_in = (period - (@last_epoch_time % period) + 1).to_i
-        ["#{prefix}:#{(@last_epoch_time / period).to_i}:#{unprefixed_key}", expires_in]
+        expires_in = period_remainder + 1
+        ["#{prefix}:#{period_number}:#{unprefixed_key}", expires_in]
+      end
+
+      def offset_for(unprefixed_key, period, use_offset)
+        use_offset ? Digest::MD5.hexdigest(unprefixed_key).hex % period : 0
+      end
+
+      def period_number_and_time_into(period, offset)
+        [((@last_epoch_time + offset) / period).to_i, (@last_epoch_time + offset) % period]
       end
 
       def do_count(key, expires_in)
diff --git a/lib/rack/attack/configuration.rb b/lib/rack/attack/configuration.rb
index a4bdc987..ada5d261 100644
--- a/lib/rack/attack/configuration.rb
+++ b/lib/rack/attack/configuration.rb
@@ -10,8 +10,7 @@ class Configuration
       DEFAULT_THROTTLED_RESPONDER = lambda do |req|
         if Rack::Attack.configuration.throttled_response_retry_after_header
           match_data = req.env['rack.attack.match_data']
-          now = match_data[:epoch_time]
-          retry_after = match_data[:period] - (now % match_data[:period])
+          retry_after = match_data[:retry_after] - match_data[:epoch_time]
 
           [429, { 'content-type' => 'text/plain', 'retry-after' => retry_after.to_s }, ["Retry later\n"]]
         else
@@ -20,7 +19,8 @@ class Configuration
       end
 
       attr_reader :safelists, :blocklists, :throttles, :anonymous_blocklists, :anonymous_safelists
-      attr_accessor :blocklisted_responder, :throttled_responder, :throttled_response_retry_after_header
+      attr_accessor :blocklisted_responder, :throttled_responder, :throttled_response_retry_after_header,
+                    :throttled_responder_is_offset_aware
 
       attr_reader :blocklisted_response, :throttled_response # Keeping these for backwards compatibility
 
@@ -91,8 +91,10 @@ def blocklisted?(request)
       end
 
       def throttled?(request)
+        use_offset = throttled_responder_is_offset_aware ||
+                     (!throttled_response && throttled_responder == DEFAULT_THROTTLED_RESPONDER)
         @throttles.any? do |_name, throttle|
-          throttle.matched_by?(request)
+          throttle.matched_by?(request, use_offset)
         end
       end
 
@@ -119,6 +121,7 @@ def set_defaults
 
         @blocklisted_responder = DEFAULT_BLOCKLISTED_RESPONDER
         @throttled_responder = DEFAULT_THROTTLED_RESPONDER
+        @throttled_responder_is_offset_aware = false
 
         # Deprecated: Keeping these for backwards compatibility
         @blocklisted_response = nil
diff --git a/lib/rack/attack/throttle.rb b/lib/rack/attack/throttle.rb
index 0ec5f7aa..51f83481 100644
--- a/lib/rack/attack/throttle.rb
+++ b/lib/rack/attack/throttle.rb
@@ -22,20 +22,21 @@ def cache
         Rack::Attack.cache
       end
 
-      def matched_by?(request)
+      def matched_by?(request, use_offset = false)
         discriminator = discriminator_for(request)
         return false unless discriminator
 
         current_period  = period_for(request)
         current_limit   = limit_for(request)
-        count           = cache.count("#{name}:#{discriminator}", current_period)
+        count           = cache.count("#{name}:#{discriminator}", current_period, use_offset)
 
         data = {
           discriminator: discriminator,
           count: count,
           period: current_period,
           limit: current_limit,
-          epoch_time: cache.last_epoch_time
+          epoch_time: cache.last_epoch_time,
+          retry_after: cache.last_retry_after_time
         }
 
         annotate_request_with_throttle_data(request, data)
diff --git a/spec/acceptance/customizing_throttled_affects_offset.rb b/spec/acceptance/customizing_throttled_affects_offset.rb
new file mode 100644
index 00000000..4e122422
--- /dev/null
+++ b/spec/acceptance/customizing_throttled_affects_offset.rb
@@ -0,0 +1,72 @@
+# frozen_string_literal: true
+
+require_relative "../spec_helper"
+require "timecop"
+
+describe "Customizing throttled response" do
+  before do
+    Rack::Attack.cache.store = ActiveSupport::Cache::MemoryStore.new
+
+    Rack::Attack.throttle("by ip", limit: 1, period: 6) do |request|
+      request.ip
+    end
+  end
+
+  it "uses offset if responder is default" do
+    assert Rack::Attack.cache.store.is_a? ActiveSupport::Cache::MemoryStore
+    assert_equal 85, count_429_responses
+  end
+
+  it "does not use offset if responder is not default and aware is not set" do
+    assert_equal false, Rack::Attack.configuration.throttled_responder_is_offset_aware
+    Rack::Attack.throttled_responder = lambda do |_req|
+      [429, {}, ["Throttled"]]
+    end
+    assert_equal 100, count_429_responses
+  end
+
+  it "uses offset if responder is not default and aware is set" do
+    Rack::Attack.throttled_responder = lambda do |_req|
+      [429, {}, ["Throttled"]]
+    end
+    Rack::Attack.configuration.throttled_responder_is_offset_aware = true
+    assert_equal 85, count_429_responses
+  end
+
+  # Count the number of responses with 429 status, out of 100 requests,
+  # when the clock advances by one second.
+  #
+  # When this is invoked with a throttle with period 6 active, using
+  # a random offset, we would expect about one in six to expire in the
+  # first second. For the fixed start_time we're using, the offset_for
+  # MD5 hash happens to come out to 15 expired, 85 throttled, out of 100.
+  # (If anything about the algorithm changes, that count probably would
+  # too.)
+  #
+  # When using an old-style period, the start_time is at the beginning of
+  # the period, since 2020-01-01 00:00:00 == 1577836800 == 262972800*6,
+  # and after 1 second we would expect 0 expires, thus 100 of 100 requests
+  # to be throttled.
+
+  def count_429_responses
+    addresses = (1..100).map { |i| "1.2.3.#{i}" }
+    start_time = Time.gm('2020-01-01 00:00:00')
+    Timecop.freeze(start_time) do
+      initial_200_response_count = 0
+      addresses.each do |ip|
+        get "/", {}, "REMOTE_ADDR" => ip
+        initial_200_response_count += 1 if last_response.status == 200
+      end
+      assert_equal 100, initial_200_response_count
+
+      final_429_response_count = 0
+      Timecop.travel(start_time + 1) do
+        addresses.each do |ip|
+          get "/", {}, "REMOTE_ADDR" => ip
+          final_429_response_count += 1 if last_response.status == 429
+        end
+      end
+      final_429_response_count
+    end
+  end
+end
diff --git a/spec/acceptance/customizing_throttled_response_spec.rb b/spec/acceptance/customizing_throttled_response_spec.rb
index 0990975e..b512801c 100644
--- a/spec/acceptance/customizing_throttled_response_spec.rb
+++ b/spec/acceptance/customizing_throttled_response_spec.rb
@@ -24,6 +24,7 @@
       [503, {}, ["Throttled"]]
     end
 
+    get "/", {}, "REMOTE_ADDR" => "1.2.3.4"
     get "/", {}, "REMOTE_ADDR" => "1.2.3.4"
 
     assert_equal 503, last_response.status
@@ -74,6 +75,7 @@
       end
     end
 
+    get "/", {}, "REMOTE_ADDR" => "1.2.3.4"
     get "/", {}, "REMOTE_ADDR" => "1.2.3.4"
 
     assert_equal 503, last_response.status
diff --git a/spec/acceptance/key_and_expiry_spec.rb b/spec/acceptance/key_and_expiry_spec.rb
new file mode 100644
index 00000000..f9890b7a
--- /dev/null
+++ b/spec/acceptance/key_and_expiry_spec.rb
@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+require_relative "../spec_helper"
+require "timecop"
+
+describe "Behavior of key_and_expiry" do
+  it "forms keys and expirations without offset as expected" do
+    unprefixed_key = "abc789"
+    period = 1000
+    time = Time.at(1_000_000_000)
+
+    Timecop.freeze(time) do
+      key, expiry = Rack::Attack.cache.send(:key_and_expiry, unprefixed_key, period, false)
+      assert_equal "rack::attack:1000000:abc789", key
+      assert_equal 1001, expiry
+    end
+  end
+
+  it "forms keys and expirations with offset as expected" do
+    unprefixed_key = "abc789"
+    period = 1000
+    time = Time.at(1_000_000_000)
+
+    Timecop.freeze(time) do
+      Rack::Attack.cache.stub :offset_for, 0 do
+        key, expiry = Rack::Attack.cache.send(:key_and_expiry, unprefixed_key, period, true)
+        assert_equal "rack::attack:1000000:abc789", key
+        assert_equal 1001, expiry
+      end
+
+      Rack::Attack.cache.stub :offset_for, 123 do
+        key, expiry = Rack::Attack.cache.send(:key_and_expiry, unprefixed_key, period, true)
+        assert_equal "rack::attack:1000000:abc789", key
+        assert_equal 1001 - 123, expiry
+      end
+
+      Digest::MD5.stub :hexdigest, "123" do
+        key, expiry = Rack::Attack.cache.send(:key_and_expiry, unprefixed_key, period, true)
+        assert_equal "rack::attack:1000000:abc789", key
+        assert_equal 1001 - "123".hex, expiry
+      end
+    end
+  end
+
+  it "expires correctly when period is 1 second" do
+    Timecop.freeze do
+      current_epoch = Time.now.to_i
+      key, expiry = Rack::Attack.cache.send(:key_and_expiry, "abc789", 1)
+      assert_equal "rack::attack:#{current_epoch}:abc789", key
+      assert_equal 2, expiry
+    end
+  end
+end
diff --git a/spec/acceptance/throttling_spec.rb b/spec/acceptance/throttling_spec.rb
index 0db89dd6..0b54d0b2 100644
--- a/spec/acceptance/throttling_spec.rb
+++ b/spec/acceptance/throttling_spec.rb
@@ -48,7 +48,19 @@
 
     Timecop.freeze(Time.at(25)) do
       get "/", {}, "REMOTE_ADDR" => "1.2.3.4"
-      assert_equal "35", last_response.headers["Retry-After"]
+      assert_equal 429, last_response.status
+      assert_equal "19", last_response.headers["Retry-After"]
+    end
+
+    Timecop.freeze(Time.at(42)) do
+      get "/", {}, "REMOTE_ADDR" => "1.2.3.4"
+      assert_equal 429, last_response.status
+      assert_equal "2", last_response.headers["Retry-After"]
+    end
+
+    Timecop.freeze(Time.at(44)) do
+      get "/", {}, "REMOTE_ADDR" => "1.2.3.4"
+      assert_equal 200, last_response.status
     end
   end
 
diff --git a/spec/allow2ban_spec.rb b/spec/allow2ban_spec.rb
index 105e0ad0..9f6533d1 100644
--- a/spec/allow2ban_spec.rb
+++ b/spec/allow2ban_spec.rb
@@ -34,7 +34,7 @@
         end
 
         it 'increases fail count' do
-          key = "rack::attack:#{Time.now.to_i / @findtime}:allow2ban:count:1.2.3.4"
+          key, _ = Rack::Attack.cache.send(:key_and_expiry, "allow2ban:count:1.2.3.4", @findtime)
 
           _(@cache.store.read(key)).must_equal 1
         end
@@ -57,7 +57,7 @@
         end
 
         it 'increases fail count' do
-          key = "rack::attack:#{Time.now.to_i / @findtime}:allow2ban:count:1.2.3.4"
+          key, _ = Rack::Attack.cache.send(:key_and_expiry, "allow2ban:count:1.2.3.4", @findtime)
           _(@cache.store.read(key)).must_equal 2
         end
 
@@ -94,7 +94,7 @@
       end
 
       it 'does not increase fail count' do
-        key = "rack::attack:#{Time.now.to_i / @findtime}:allow2ban:count:1.2.3.4"
+        key, _ = Rack::Attack.cache.send(:key_and_expiry, "allow2ban:count:1.2.3.4", @findtime)
         _(@cache.store.read(key)).must_equal 2
       end
 
@@ -114,7 +114,7 @@
       end
 
       it 'does not increase fail count' do
-        key = "rack::attack:#{Time.now.to_i / @findtime}:allow2ban:count:1.2.3.4"
+        key, _ = Rack::Attack.cache.send(:key_and_expiry, "allow2ban:count:1.2.3.4", @findtime)
         _(@cache.store.read(key)).must_equal 2
       end
 
diff --git a/spec/fail2ban_spec.rb b/spec/fail2ban_spec.rb
index 6a9d9bcf..42a0d2b5 100644
--- a/spec/fail2ban_spec.rb
+++ b/spec/fail2ban_spec.rb
@@ -33,7 +33,7 @@
         end
 
         it 'increases fail count' do
-          key = "rack::attack:#{Time.now.to_i / @findtime}:fail2ban:count:1.2.3.4"
+          key, _ = Rack::Attack.cache.send(:key_and_expiry, "fail2ban:count:1.2.3.4", @findtime)
           _(@cache.store.read(key)).must_equal 1
         end
 
@@ -55,7 +55,7 @@
         end
 
         it 'increases fail count' do
-          key = "rack::attack:#{Time.now.to_i / @findtime}:fail2ban:count:1.2.3.4"
+          key, _ = Rack::Attack.cache.send(:key_and_expiry, "fail2ban:count:1.2.3.4", @findtime)
           _(@cache.store.read(key)).must_equal 2
         end
 
@@ -77,7 +77,7 @@
         end
 
         it 'resets fail count' do
-          key = "rack::attack:#{Time.now.to_i / @findtime}:fail2ban:count:1.2.3.4"
+          key, _ = Rack::Attack.cache.send(:key_and_expiry, "fail2ban:count:1.2.3.4", @findtime)
           assert_nil @cache.store.read(key)
         end
 
@@ -113,7 +113,7 @@
       end
 
       it 'does not increase fail count' do
-        key = "rack::attack:#{Time.now.to_i / @findtime}:fail2ban:count:1.2.3.4"
+        key, _ = Rack::Attack.cache.send(:key_and_expiry, "fail2ban:count:1.2.3.4", @findtime)
         _(@cache.store.read(key)).must_equal 2
       end
 
@@ -133,7 +133,7 @@
       end
 
       it 'does not increase fail count' do
-        key = "rack::attack:#{Time.now.to_i / @findtime}:fail2ban:count:1.2.3.4"
+        key, _ = Rack::Attack.cache.send(:key_and_expiry, "fail2ban:count:1.2.3.4", @findtime)
         _(@cache.store.read(key)).must_equal 2
       end
 
diff --git a/spec/rack_attack_throttle_spec.rb b/spec/rack_attack_throttle_spec.rb
index 0b0d68ac..e01fafa0 100644
--- a/spec/rack_attack_throttle_spec.rb
+++ b/spec/rack_attack_throttle_spec.rb
@@ -17,7 +17,7 @@
     before { get '/', {}, 'REMOTE_ADDR' => '1.2.3.4' }
 
     it 'should set the counter for one request' do
-      key = "rack::attack:#{Time.now.to_i / @period}:ip/sec:1.2.3.4"
+      key, _ = Rack::Attack.cache.send(:key_and_expiry, "ip/sec:1.2.3.4", @period, true)
       _(Rack::Attack.cache.store.read(key)).must_equal 1
     end
 
@@ -27,6 +27,7 @@
         limit: 1,
         period: @period,
         epoch_time: Rack::Attack.cache.last_epoch_time.to_i,
+        retry_after: Rack::Attack.cache.last_retry_after_time.to_i,
         discriminator: "1.2.3.4"
       }
 
@@ -52,6 +53,7 @@
         limit: 1,
         period: @period,
         epoch_time: Rack::Attack.cache.last_epoch_time.to_i,
+        retry_after: Rack::Attack.cache.last_retry_after_time.to_i,
         discriminator: "1.2.3.4"
       )
 
@@ -73,7 +75,7 @@
     before { get '/', {}, 'REMOTE_ADDR' => '1.2.3.4' }
 
     it 'should set the counter for one request' do
-      key = "rack::attack:#{Time.now.to_i / @period}:ip/sec:1.2.3.4"
+      key, _ = Rack::Attack.cache.send(:key_and_expiry, "ip/sec:1.2.3.4", @period, true)
       _(Rack::Attack.cache.store.read(key)).must_equal 1
     end
 
@@ -83,6 +85,7 @@
         limit: 1,
         period: @period,
         epoch_time: Rack::Attack.cache.last_epoch_time.to_i,
+        retry_after: Rack::Attack.cache.last_retry_after_time.to_i,
         discriminator: "1.2.3.4"
       }
 
@@ -104,7 +107,7 @@
     before { get '/', {}, 'REMOTE_ADDR' => '1.2.3.4' }
 
     it 'should set the counter for one request' do
-      key = "rack::attack:#{Time.now.to_i / @period}:ip/sec:1.2.3.4"
+      key, _ = Rack::Attack.cache.send(:key_and_expiry, "ip/sec:1.2.3.4", @period, true)
       _(Rack::Attack.cache.store.read(key)).must_equal 1
     end
 
@@ -114,6 +117,7 @@
         limit: 1,
         period: @period,
         epoch_time: Rack::Attack.cache.last_epoch_time.to_i,
+        retry_after: Rack::Attack.cache.last_retry_after_time.to_i,
         discriminator: "1.2.3.4"
       }
 
@@ -135,7 +139,7 @@
     before { get '/', {}, 'REMOTE_ADDR' => '1.2.3.4' }
 
     it 'should not set the counter' do
-      key = "rack::attack:#{Time.now.to_i / @period}:ip/sec:1.2.3.4"
+      key, _ = Rack::Attack.cache.send(:key_and_expiry, "ip/sec:1.2.3.4", @period, true)
       assert_nil Rack::Attack.cache.store.read(key)
     end
 
@@ -163,7 +167,7 @@
 
   it 'should not differentiate requests when throttle_discriminator_normalizer is enabled' do
     post_logins
-    key = "rack::attack:#{Time.now.to_i / @period}:logins/email:person@example.com"
+    key, _ = Rack::Attack.cache.send(:key_and_expiry, "logins/email:person@example.com", @period, true)
     _(Rack::Attack.cache.store.read(key)).must_equal 3
   end
 
@@ -174,7 +178,7 @@
 
       post_logins
       @emails.each do |email|
-        key = "rack::attack:#{Time.now.to_i / @period}:logins/email:#{email}"
+        key, _ = Rack::Attack.cache.send(:key_and_expiry, "logins/email:#{email}", @period, true)
         _(Rack::Attack.cache.store.read(key)).must_equal 1
       end
     ensure