diff --git a/History.md b/History.md
index a92a6b11..eecddf4e 100644
--- a/History.md
+++ b/History.md
@@ -5,6 +5,7 @@ Dalli Changelog
 ==========
 
 - Support rcvbuff and sndbuff byte configuration. (btatnall)
+- Add `:cache_nils` option to support nil values in `DalliStore#fetch` and `Dalli::Client#fetch` (wjordan, #559)
 
 2.7.4
 ==========
diff --git a/README.md b/README.md
index 07fcaca5..6eefd455 100644
--- a/README.md
+++ b/README.md
@@ -188,7 +188,9 @@ If serving compressed data using nginx's HttpMemcachedModule, set `memcached_gzi
 
 **sndbuf**: In bytes, set the socket SO_SNDBUF. Defaults to operating system default.
 
-**rcvbuf**: In bytes, set the socket SO_RCVBUF. Defatuls to operating system default.
+**rcvbuf**: In bytes, set the socket SO_RCVBUF. Defaults to operating system default.
+
+**cache_nils**: Boolean. If true Dalli will not treat cached `nil` values as 'not found' for `#fetch` operations. Default is false.
 
 Features and Changes
 ------------------------
diff --git a/lib/active_support/cache/dalli_store.rb b/lib/active_support/cache/dalli_store.rb
index 9620cd0f..fdbc32e4 100644
--- a/lib/active_support/cache/dalli_store.rb
+++ b/lib/active_support/cache/dalli_store.rb
@@ -82,31 +82,41 @@ def with(&block)
         @data.with(&block)
       end
 
+      # Fetch the value associated with the key.
+      # If a value is found, then it is returned.
+      #
+      # If a value is not found and no block is given, then nil is returned.
+      #
+      # If a value is not found (or if the found value is nil and :cache_nils is false)
+      # and a block is given, the block will be invoked and its return value
+      # written to the cache and returned.
       def fetch(name, options=nil)
         options ||= {}
+        options[:cache_nils] = true if @options[:cache_nils]
         namespaced_name = namespaced_key(name, options)
-
+        not_found = options[:cache_nils] ? Dalli::Server::NOT_FOUND : nil
         if block_given?
+          entry = not_found
           unless options[:force]
             entry = instrument(:read, namespaced_name, options) do |payload|
               read_entry(namespaced_name, options).tap do |result|
                 if payload
                   payload[:super_operation] = :fetch
-                  payload[:hit] = !result.nil?
+                  payload[:hit] = result != not_found
                 end
               end
             end
           end
 
-          if !entry.nil?
-            instrument(:fetch_hit, name, options) { |payload| }
-            entry
-          else
+          if entry == not_found
             result = instrument(:generate, name, options) do |payload|
               yield
             end
             write(name, result, options)
             result
+          else
+            instrument(:fetch_hit, name, options) { |payload|}
+            entry
           end
         else
           read(name, options)
diff --git a/lib/dalli/client.rb b/lib/dalli/client.rb
index 45af9f78..a6d5c281 100644
--- a/lib/dalli/client.rb
+++ b/lib/dalli/client.rb
@@ -27,6 +27,7 @@ class Client
     # - :compress - defaults to false, if true Dalli will compress values larger than 1024 bytes before sending them to memcached.
     # - :serializer - defaults to Marshal
     # - :compressor - defaults to zlib
+    # - :cache_nils - defaults to false, if true Dalli will not treat cached nil values as 'not found' for #fetch operations.
     #
     def initialize(servers=nil, options={})
       @servers = normalize_servers(servers || ENV["MEMCACHE_SERVERS"] || '127.0.0.1:11211')
@@ -52,8 +53,9 @@ def multi
 
     ##
     # Get the value associated with the key.
+    # If a value is not found, then +nil+ is returned.
     def get(key, options=nil)
-      perform(:get, key)
+      perform(:get, key, options)
     end
 
     ##
@@ -71,10 +73,24 @@ def get_multi(*keys)
       end
     end
 
+    CACHE_NILS = {cache_nils: true}.freeze
+
+    # Fetch the value associated with the key.
+    # If a value is found, then it is returned.
+    #
+    # If a value is not found and no block is given, then nil is returned.
+    #
+    # If a value is not found (or if the found value is nil and :cache_nils is false)
+    # and a block is given, the block will be invoked and its return value
+    # written to the cache and returned.
     def fetch(key, ttl=nil, options=nil)
       ttl ||= @options[:expires_in].to_i
+      options = options.nil? ? CACHE_NILS : options.merge(CACHE_NILS) if @options[:cache_nils]
       val = get(key, options)
-      if val.nil? && block_given?
+      not_found = @options[:cache_nils] ?
+        val == Dalli::Server::NOT_FOUND :
+        val.nil?
+      if not_found && block_given?
         val = yield
         add(key, val, ttl, options)
       end
diff --git a/lib/dalli/server.rb b/lib/dalli/server.rb
index f5900ec9..f2d6eb95 100644
--- a/lib/dalli/server.rb
+++ b/lib/dalli/server.rb
@@ -267,10 +267,10 @@ def multi?
       Thread.current[:dalli_multi]
     end
 
-    def get(key)
+    def get(key, options=nil)
       req = [REQUEST, OPCODES[:get], key.bytesize, 0, 0, 0, key.bytesize, 0, 0, key].pack(FORMAT[:get])
       write(req)
-      generic_response(true)
+      generic_response(true, !!(options && options.is_a?(Hash) && options[:cache_nils]))
     end
 
     def send_multiget(keys)
@@ -492,11 +492,15 @@ def sanitize_ttl(ttl)
       end
     end
 
-    def generic_response(unpack=false)
+    # Implements the NullObject pattern to store an application-defined value for 'Key not found' responses.
+    class NilObject; end
+    NOT_FOUND = NilObject.new
+
+    def generic_response(unpack=false, cache_nils=false)
       (extras, _, status, count) = read_header.unpack(NORMAL_HEADER)
       data = read(count) if count > 0
       if status == 1
-        nil
+        cache_nils ? NOT_FOUND : nil
       elsif status == 2 || status == 5
         false # Not stored, normal status for add operation
       elsif status != 0
diff --git a/test/helper.rb b/test/helper.rb
index c75cceca..54042eee 100644
--- a/test/helper.rb
+++ b/test/helper.rb
@@ -48,6 +48,16 @@ def with_activesupport
     yield
   end
 
+  def with_activesupport_memcached
+    port = 19987
+    with_activesupport do
+      memcached_persistent(port) do
+        connect(port)
+        yield
+      end
+    end
+  end
+
   def with_actionpack
     require 'action_dispatch'
     require 'action_controller'
diff --git a/test/memcached_mock.rb b/test/memcached_mock.rb
index 3df6ffb9..7ecc2953 100644
--- a/test/memcached_mock.rb
+++ b/test/memcached_mock.rb
@@ -78,8 +78,8 @@ def find_memcached
       raise Errno::ENOENT, "Unable to find memcached 1.4+ locally"
     end
 
-    def memcached_persistent(port=21345)
-      dc = start_and_flush_with_retry(port, '', {})
+    def memcached_persistent(port=21345, options={})
+      dc = start_and_flush_with_retry(port, '', options)
       yield dc, port if block_given?
     end
 
diff --git a/test/test_active_support.rb b/test/test_active_support.rb
index 2ae88939..b6263bde 100644
--- a/test/test_active_support.rb
+++ b/test/test_active_support.rb
@@ -47,25 +47,64 @@ def cache_key
       end
     end
 
-    it 'support fetch' do
-      with_activesupport do
-        memcached_persistent(@port) do
-          connect(@port)
-          dvalue = @dalli.fetch('someotherkeywithoutspaces', :expires_in => 1.second) { 123 }
-          assert_equal 123, dvalue
+    describe 'fetch' do
+      it 'support expires_in' do
+        with_activesupport_memcached do
+            dvalue = @dalli.fetch('someotherkeywithoutspaces', :expires_in => 1.second) { 123 }
+            assert_equal 123, dvalue
+        end
+      end
 
+      it 'support object' do
+        with_activesupport_memcached do
           o = Object.new
           o.instance_variable_set :@foo, 'bar'
-          dvalue = @dalli.fetch(rand_key, :raw => true) { o }
+          dvalue = @dalli.fetch(rand_key) { o }
           assert_equal o, dvalue
+        end
+      end
 
-          dvalue = @dalli.fetch(rand_key) { o }
+      it 'support object with raw' do
+        with_activesupport_memcached do
+          o = Object.new
+          o.instance_variable_set :@foo, 'bar'
+          dvalue = @dalli.fetch(rand_key, :raw => true) { o }
           assert_equal o, dvalue
+        end
+      end
 
+      it 'support false value' do
+        with_activesupport_memcached do
           @dalli.write('false', false)
           dvalue = @dalli.fetch('false') { flunk }
           assert_equal false, dvalue
+        end
+      end
+
+      it 'support nil value when cache_nils: true' do
+        @dalli = ActiveSupport::Cache.lookup_store(:dalli_store, "localhost:#{@port}", cache_nils: true)
+        @dalli.clear
+
+        memcached_persistent(@port) do
+          @dalli.write('nil', nil)
+          dvalue = @dalli.fetch('nil') { flunk }
+          assert_equal nil, dvalue
+        end
+
+        @dalli = ActiveSupport::Cache.lookup_store(:dalli_store, "localhost:#{@port}", cache_nils: false)
+        @dalli.clear
+
+        memcached_persistent(@port) do
+          @dalli.write('nil', nil)
+          executed = false
+          dvalue = @dalli.fetch('nil') { executed = true; 'bar' }
+          assert_equal true, executed
+          assert_equal 'bar', dvalue
+        end
+      end
 
+      it 'support object with cache_key' do
+        with_activesupport_memcached do
           user = MockUser.new
           @dalli.write(user.cache_key, false)
           dvalue = @dalli.fetch(user) { flunk }
diff --git a/test/test_dalli.rb b/test/test_dalli.rb
index fce4e305..fea66117 100644
--- a/test/test_dalli.rb
+++ b/test/test_dalli.rb
@@ -216,10 +216,25 @@
         dc.set("fetch_key", false)
         res = dc.fetch("fetch_key") { flunk "fetch block called" }
         assert_equal false, res
+      end
+    end
+
+    it "support the fetch operation with nil values when cache_nils: true" do
+      memcached_persistent(21345, cache_nils: true) do |dc|
+        dc.flush
 
         dc.set("fetch_key", nil)
-        res = dc.fetch("fetch_key") { "bob" }
-        assert_equal 'bob', res
+        res = dc.fetch("fetch_key") { flunk "fetch block called" }
+        assert_equal nil, res
+      end
+
+      memcached_persistent(21345, cache_nils: false) do |dc|
+        dc.flush
+        dc.set("fetch_key", nil)
+        executed = false
+        res = dc.fetch("fetch_key") { executed = true; 'bar' }
+        assert_equal 'bar', res
+        assert_equal true, executed
       end
     end