diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index e0ed85eb8f..abe7d67463 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -354,6 +354,7 @@ def lock(
         name: KeyT,
         timeout: Optional[float] = None,
         sleep: float = 0.1,
+        blocking: bool = True,
         blocking_timeout: Optional[float] = None,
         lock_class: Optional[Type[Lock]] = None,
         thread_local: bool = True,
@@ -369,6 +370,12 @@ def lock(
         when the lock is in blocking mode and another client is currently
         holding the lock.
 
+        ``blocking`` indicates whether calling ``acquire`` should block until
+        the lock has been acquired or to fail immediately, causing ``acquire``
+        to return False and the lock not being acquired. Defaults to True.
+        Note this value can be overridden by passing a ``blocking``
+        argument to ``acquire``.
+
         ``blocking_timeout`` indicates the maximum amount of time in seconds to
         spend trying to acquire the lock. A value of ``None`` indicates
         continue trying forever. ``blocking_timeout`` can be specified as a
@@ -411,6 +418,7 @@ def lock(
             name,
             timeout=timeout,
             sleep=sleep,
+            blocking=blocking,
             blocking_timeout=blocking_timeout,
             thread_local=thread_local,
         )
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index d5a38b2878..0e9a7d3ee8 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -793,6 +793,7 @@ def lock(
         name: KeyT,
         timeout: Optional[float] = None,
         sleep: float = 0.1,
+        blocking: bool = True,
         blocking_timeout: Optional[float] = None,
         lock_class: Optional[Type[Lock]] = None,
         thread_local: bool = True,
@@ -808,6 +809,12 @@ def lock(
         when the lock is in blocking mode and another client is currently
         holding the lock.
 
+        ``blocking`` indicates whether calling ``acquire`` should block until
+        the lock has been acquired or to fail immediately, causing ``acquire``
+        to return False and the lock not being acquired. Defaults to True.
+        Note this value can be overridden by passing a ``blocking``
+        argument to ``acquire``.
+
         ``blocking_timeout`` indicates the maximum amount of time in seconds to
         spend trying to acquire the lock. A value of ``None`` indicates
         continue trying forever. ``blocking_timeout`` can be specified as a
@@ -850,6 +857,7 @@ def lock(
             name,
             timeout=timeout,
             sleep=sleep,
+            blocking=blocking,
             blocking_timeout=blocking_timeout,
             thread_local=thread_local,
         )
diff --git a/tests/test_asyncio/test_lock.py b/tests/test_asyncio/test_lock.py
index 56387fa954..15c3ec5162 100644
--- a/tests/test_asyncio/test_lock.py
+++ b/tests/test_asyncio/test_lock.py
@@ -97,6 +97,14 @@ async def test_float_timeout(self, r):
         assert 8 < (await r.pttl("foo")) <= 9500
         await lock.release()
 
+    async def test_blocking(self, r):
+        blocking = False
+        lock = self.get_lock(r, "foo", blocking=blocking)
+        assert not lock.blocking
+
+        lock_2 = self.get_lock(r, "foo")
+        assert lock_2.blocking
+
     async def test_blocking_timeout(self, r, event_loop):
         lock1 = self.get_lock(r, "foo")
         assert await lock1.acquire(blocking=False)