diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 51c267c95f..22c5030e6c 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -207,9 +207,6 @@ def __repr__(self):
         repr_args = ",".join((f"{k}={v}" for k, v in self.repr_pieces()))
         return f"{self.__class__.__name__}<{repr_args}>"
 
-    def __del__(self):
-        self._close_socket()
-
     @abstractmethod
     def repr_pieces(self):
         pass
@@ -380,26 +377,6 @@ async def disconnect(self, nowait: bool = False) -> None:
                 f"Timed out closing connection after {self.socket_connect_timeout}"
             ) from None
 
-    def _close_socket(self):
-        """Close the socket directly.  Used during garbage collection to
-        make sure the underlying socket is released.  This does not happen
-        reliably when the stream is garbage collected.  This is a safety
-        precaution, correct use of the library should ensure that
-        sockets are disconnected properly.
-        """
-        # some test classes don't even have this
-        writer = getattr(self, "_writer", None)
-        if writer:
-            if os.getpid() == self.pid:
-                try:
-                    writer.close()
-                except RuntimeError:
-                    # This may fail if the event loop is already closed,
-                    # even though this is not an async call.  In this
-                    # case, just ignore the error, since it is during
-                    # exit anyway.
-                    pass
-
     async def _send_ping(self):
         """Send PING, expect PONG in return"""
         await self.send_command("PING", check_health=False)
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index c4816a7e53..7c177a53cd 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -320,32 +320,3 @@ async def get_redis_connection():
     assert r1.auto_close_connection_pool is False
     await r1.connection_pool.disconnect()
     await r1.close()
-
-
-@pytest.mark.onlynoncluster
-@pytest.mark.parametrize("from_url", (True, False))
-async def test_connection_socket_cleanup(request, from_url):
-    """Verify that connections are cleaned up when they
-    are garbage collected
-    """
-    if platform.python_implementation() != "CPython":
-        pytest.skip("only works on CPython")
-    url: str = request.config.getoption("--redis-url")
-    url_args = parse_url(url)
-
-    async def get_redis_connection():
-        if from_url:
-            return Redis.from_url(url)
-        return Redis(**url_args)
-
-    async def do_something(redis):
-        await redis.incr("counter")
-        await redis.close()
-
-    mock = Mock()
-    with patch.object(AbstractConnection, "_close_socket", mock):
-        r1 = await get_redis_connection()
-        await do_something(r1)
-        r1 = None
-
-    assert mock.call_count == 1