diff --git a/.rubocop.yml b/.rubocop.yml
index 9c2e0b0b2..b51594620 100644
--- a/.rubocop.yml
+++ b/.rubocop.yml
@@ -132,3 +132,7 @@ Metrics/BlockNesting:
 
 Style/HashTransformValues:
   Enabled: false
+
+Style/SymbolProc:
+  Exclude:
+    - 'test/**/*'
\ No newline at end of file
diff --git a/README.md b/README.md
index b8591ca50..a83c3e0fc 100644
--- a/README.md
+++ b/README.md
@@ -184,9 +184,9 @@ commands to Redis and gathers their replies. These replies are returned
 by the `#pipelined` method.
 
 ```ruby
-redis.pipelined do
-  redis.set "foo", "bar"
-  redis.incr "baz"
+redis.pipelined do |pipeline|
+  pipeline.set "foo", "bar"
+  pipeline.incr "baz"
 end
 # => ["OK", 1]
 ```
@@ -210,15 +210,15 @@ end
 ### Futures
 
 Replies to commands in a pipeline can be accessed via the *futures* they
-emit (since redis-rb 3.0). All calls inside a pipeline block return a
+emit (since redis-rb 3.0). All calls on the pipeline object return a
 `Future` object, which responds to the `#value` method. When the
 pipeline has successfully executed, all futures are assigned their
 respective replies and can be used.
 
 ```ruby
-redis.pipelined do
-  @set = redis.set "foo", "bar"
-  @incr = redis.incr "baz"
+redis.pipelined do |pipeline|
+  @set = pipeline.set "foo", "bar"
+  @incr = pipeline.incr "baz"
 end
 
 @set.value
diff --git a/lib/redis.rb b/lib/redis.rb
index f96a94e55..b5caf076b 100644
--- a/lib/redis.rb
+++ b/lib/redis.rb
@@ -5,6 +5,7 @@
 require "redis/commands"
 
 class Redis
+  BASE_PATH = __dir__
   @exists_returns_integer = true
 
   class << self
@@ -104,6 +105,11 @@ def close
   # See http://redis.io/topics/pipelining for more details.
   #
   def queue(*command)
+    Kernel.warn(
+      "Redis#queue is deprecated and will be removed in Redis 5.0.0. Use Redis#pipelined instead." \
+      "(called from: #{caller(1, 1).first})"
+    )
+
     synchronize do
       @queue[Thread.current.object_id] << command
     end
@@ -114,6 +120,11 @@ def queue(*command)
   # See http://redis.io/topics/pipelining for more details.
   #
   def commit
+    Kernel.warn(
+      "Redis#commit is deprecated and will be removed in Redis 5.0.0. Use Redis#pipelined instead. " \
+      "(called from: #{Kernel.caller(1, 1).first})"
+    )
+
     synchronize do |client|
       begin
         pipeline = Pipeline.new(client)
@@ -193,12 +204,20 @@ def unwatch
     end
   end
 
-  def pipelined
+  def pipelined(&block)
+    deprecation_displayed = false
+    if block&.arity == 0
+      Pipeline.deprecation_warning(Kernel.caller_locations(1, 5))
+      deprecation_displayed = true
+    end
+
     synchronize do |prior_client|
       begin
-        @client = Pipeline.new(prior_client)
-        yield(self)
-        prior_client.call_pipeline(@client)
+        pipeline = Pipeline.new(prior_client)
+        @client = deprecation_displayed ? pipeline : DeprecatedPipeline.new(pipeline)
+        pipelined_connection = PipelinedConnection.new(pipeline)
+        yield pipelined_connection
+        prior_client.call_pipeline(pipeline)
       ensure
         @client = prior_client
       end
diff --git a/lib/redis/pipeline.rb b/lib/redis/pipeline.rb
index c6a6dc382..bc78a574b 100644
--- a/lib/redis/pipeline.rb
+++ b/lib/redis/pipeline.rb
@@ -1,7 +1,70 @@
 # frozen_string_literal: true
 
+require "delegate"
+
 class Redis
+  class PipelinedConnection
+    def initialize(pipeline)
+      @pipeline = pipeline
+    end
+
+    include Commands
+
+    def db
+      @pipeline.db
+    end
+
+    def db=(db)
+      @pipeline.db = db
+    end
+
+    def pipelined
+      yield self
+    end
+
+    private
+
+    def synchronize
+      yield self
+    end
+
+    def send_command(command, &block)
+      @pipeline.call(command, &block)
+    end
+
+    def send_blocking_command(command, timeout, &block)
+      @pipeline.call_with_timeout(command, timeout, &block)
+    end
+  end
+
   class Pipeline
+    REDIS_INTERNAL_PATH = File.expand_path("..", __dir__).freeze
+    # Redis use MonitorMixin#synchronize and this class use DelegateClass which we want to filter out.
+    # Both are in the stdlib so we can simply filter the entire stdlib out.
+    STDLIB_PATH = File.expand_path("..", MonitorMixin.instance_method(:synchronize).source_location.first).freeze
+
+    class << self
+      def deprecation_warning(caller_locations) # :nodoc:
+        callsite = caller_locations.find { |l| !l.path.start_with?(REDIS_INTERNAL_PATH, STDLIB_PATH) }
+        callsite ||= caller_locations.last # The caller_locations should be large enough, but just in case.
+        Kernel.warn <<~MESSAGE
+          Pipelining commands on a Redis instance is deprecated and will be removed in Redis 5.0.0.
+
+          redis.pipelined do
+            redis.get("key")
+          end
+
+          should be replaced by
+
+          redis.pipelined do |pipeline|
+            pipeline.get("key")
+          end
+
+          (called from #{callsite}}
+        MESSAGE
+      end
+    end
+
     attr_accessor :db
     attr_reader :client
 
@@ -124,6 +187,21 @@ def commands
     end
   end
 
+  DeprecatedPipeline = DelegateClass(Pipeline) do
+    def initialize(pipeline)
+      super(pipeline)
+      @deprecation_displayed = false
+    end
+
+    def __getobj__
+      unless @deprecation_displayed
+        Pipeline.deprecation_warning(Kernel.caller_locations(1, 10))
+        @deprecation_displayed = true
+      end
+      @delegate_dc_obj
+    end
+  end
+
   class FutureNotReady < RuntimeError
     def initialize
       super("Value will be available once the pipeline executes.")
diff --git a/test/pipelining_commands_test.rb b/test/pipelining_commands_test.rb
index be5243b85..394bbbb79 100644
--- a/test/pipelining_commands_test.rb
+++ b/test/pipelining_commands_test.rb
@@ -6,9 +6,9 @@ class TestPipeliningCommands < Minitest::Test
   include Helper::Client
 
   def test_bulk_commands
-    r.pipelined do
-      r.lpush "foo", "s1"
-      r.lpush "foo", "s2"
+    r.pipelined do |p|
+      p.lpush "foo", "s1"
+      p.lpush "foo", "s2"
     end
 
     assert_equal 2, r.llen("foo")
@@ -17,9 +17,9 @@ def test_bulk_commands
   end
 
   def test_multi_bulk_commands
-    r.pipelined do
-      r.mset("foo", "s1", "bar", "s2")
-      r.mset("baz", "s3", "qux", "s4")
+    r.pipelined do |p|
+      p.mset("foo", "s1", "bar", "s2")
+      p.mset("baz", "s3", "qux", "s4")
     end
 
     assert_equal "s1", r.get("foo")
@@ -29,10 +29,10 @@ def test_multi_bulk_commands
   end
 
   def test_bulk_and_multi_bulk_commands_mixed
-    r.pipelined do
-      r.lpush "foo", "s1"
-      r.lpush "foo", "s2"
-      r.mset("baz", "s3", "qux", "s4")
+    r.pipelined do |p|
+      p.lpush "foo", "s1"
+      p.lpush "foo", "s2"
+      p.mset("baz", "s3", "qux", "s4")
     end
 
     assert_equal 2, r.llen("foo")
@@ -43,10 +43,10 @@ def test_bulk_and_multi_bulk_commands_mixed
   end
 
   def test_multi_bulk_and_bulk_commands_mixed
-    r.pipelined do
-      r.mset("baz", "s3", "qux", "s4")
-      r.lpush "foo", "s1"
-      r.lpush "foo", "s2"
+    r.pipelined do |p|
+      p.mset("baz", "s3", "qux", "s4")
+      p.lpush "foo", "s1"
+      p.lpush "foo", "s2"
     end
 
     assert_equal 2, r.llen("foo")
@@ -64,19 +64,19 @@ def test_pipelined_with_an_empty_block
   end
 
   def test_returning_the_result_of_a_pipeline
-    result = r.pipelined do
-      r.set "foo", "bar"
-      r.get "foo"
-      r.get "bar"
+    result = r.pipelined do |p|
+      p.set "foo", "bar"
+      p.get "foo"
+      p.get "bar"
     end
 
     assert_equal ["OK", "bar", nil], result
   end
 
   def test_assignment_of_results_inside_the_block
-    r.pipelined do
-      @first = r.sadd("foo", 1)
-      @second = r.sadd("foo", 1)
+    r.pipelined do |p|
+      @first = p.sadd("foo", 1)
+      @second = p.sadd("foo", 1)
     end
 
     assert_equal true, @first.value
@@ -87,10 +87,10 @@ def test_assignment_of_results_inside_the_block
   # it doesn't make a lot of sense.
   def test_assignment_of_results_inside_the_block_with_errors
     assert_raises(Redis::CommandError) do
-      r.pipelined do
-        r.doesnt_exist
-        @first = r.sadd("foo", 1)
-        @second = r.sadd("foo", 1)
+      r.pipelined do |p|
+        p.doesnt_exist
+        @first = p.sadd("foo", 1)
+        @second = p.sadd("foo", 1)
       end
     end
 
@@ -99,11 +99,11 @@ def test_assignment_of_results_inside_the_block_with_errors
   end
 
   def test_assignment_of_results_inside_a_nested_block
-    r.pipelined do
-      @first = r.sadd("foo", 1)
+    r.pipelined do |p|
+      @first = p.sadd("foo", 1)
 
-      r.pipelined do
-        @second = r.sadd("foo", 1)
+      r.pipelined do |p2|
+        @second = p2.sadd("foo", 1)
       end
     end
 
@@ -112,32 +112,32 @@ def test_assignment_of_results_inside_a_nested_block
   end
 
   def test_futures_raise_when_confused_with_something_else
-    r.pipelined do
-      @result = r.sadd("foo", 1)
+    r.pipelined do |p|
+      @result = p.sadd("foo", 1)
     end
 
     assert_raises(NoMethodError) { @result.to_s }
   end
 
   def test_futures_raise_when_trying_to_access_their_values_too_early
-    r.pipelined do
+    r.pipelined do |p|
       assert_raises(Redis::FutureNotReady) do
-        r.sadd("foo", 1).value
+        p.sadd("foo", 1).value
       end
     end
   end
 
   def test_futures_raise_when_command_errors_and_needs_transformation
     assert_raises(Redis::CommandError) do
-      r.pipelined do
-        @result = r.zrange("a", "b", 5, with_scores: true)
+      r.pipelined do |p|
+        @result = p.zrange("a", "b", 5, with_scores: true)
       end
     end
   end
 
   def test_futures_warn_when_tested_for_equality
-    r.pipelined do
-      @result = r.sadd("foo", 1)
+    r.pipelined do |p|
+      @result = p.sadd("foo", 1)
     end
 
     assert_output(nil, /deprecated/) do
@@ -146,8 +146,8 @@ def test_futures_warn_when_tested_for_equality
   end
 
   def test_futures_can_be_identified
-    r.pipelined do
-      @result = r.sadd("foo", 1)
+    r.pipelined do |p|
+      @result = p.sadd("foo", 1)
     end
 
     assert_equal true, @result.is_a?(Redis::Future)
@@ -163,10 +163,10 @@ def test_returning_the_result_of_an_empty_pipeline
   end
 
   def test_nesting_pipeline_blocks
-    r.pipelined do
-      r.set("foo", "s1")
-      r.pipelined do
-        r.set("bar", "s2")
+    r.pipelined do |p|
+      p.set("foo", "s1")
+      p.pipelined do |p2|
+        p2.set("bar", "s2")
       end
     end
 
@@ -175,16 +175,16 @@ def test_nesting_pipeline_blocks
   end
 
   def test_info_in_a_pipeline_returns_hash
-    result = r.pipelined do
-      r.info
+    result = r.pipelined do |p|
+      p.info
     end
 
     assert result.first.is_a?(Hash)
   end
 
   def test_config_get_in_a_pipeline_returns_hash
-    result = r.pipelined do
-      r.config(:get, "*")
+    result = r.pipelined do |p|
+      p.config(:get, "*")
     end
 
     assert result.first.is_a?(Hash)
@@ -192,8 +192,8 @@ def test_config_get_in_a_pipeline_returns_hash
 
   def test_hgetall_in_a_pipeline_returns_hash
     r.hmset("hash", "field", "value")
-    result = r.pipelined do
-      r.hgetall("hash")
+    result = r.pipelined do |p|
+      p.hgetall("hash")
     end
 
     assert_equal result.first, { "field" => "value" }
@@ -201,8 +201,8 @@ def test_hgetall_in_a_pipeline_returns_hash
 
   def test_keys_in_a_pipeline
     r.set("key", "value")
-    result = r.pipelined do
-      r.keys("*")
+    result = r.pipelined do |p|
+      p.keys("*")
     end
 
     assert_equal ["key"], result.first