diff --git a/synapse/replication/tcp/handler.py b/synapse/replication/tcp/handler.py
index b94bdcfd1a..5e00e312df 100644
--- a/synapse/replication/tcp/handler.py
+++ b/synapse/replication/tcp/handler.py
@@ -274,7 +274,7 @@ class ReplicationCommandHandler:
         self._command_queues_by_stream = {
             stream_name: BackgroundQueue[_StreamCommandQueueItem](
                 hs,
-                "process-replication-data",
+                f"process-replication-data-rei{stream_name}",
                 self._unsafe_process_item,
             )
             for stream_name in self._streams
diff --git a/synapse/replication/tcp/redis.py b/synapse/replication/tcp/redis.py
index d042d82ca4..993fed85eb 100644
--- a/synapse/replication/tcp/redis.py
+++ b/synapse/replication/tcp/redis.py
@@ -224,8 +224,12 @@ class RedisSubscriber(SubscriberProtocol):
 
         if isawaitable(res):
 
-            async def wrapper():
-                with Measure(f"reireplication_on_{cmd.NAME}"):
+            async def wrapper() -> None:
+                with Measure(
+                    self.hs.get_clock(),
+                    name=f"reireplication_on_{cmd.NAME}",
+                    server_name=self.server_name,
+                ):
                     return await res
 
             self.hs.run_as_background_process(
diff --git a/synapse/util/background_queue.py b/synapse/util/background_queue.py
index dfea7247f4..9593076e02 100644
--- a/synapse/util/background_queue.py
+++ b/synapse/util/background_queue.py
@@ -28,6 +28,7 @@ from twisted.internet import defer
 
 from synapse.util.async_helpers import DeferredEvent
 from synapse.util.duration import Duration
+from synapse.util.metrics import Measure
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -65,6 +66,7 @@ class BackgroundQueue(Generic[T]):
         timeout_ms: int = 1000,
     ) -> None:
         self._hs = hs
+        self._clock = hs.get_clock()
         self._name = name
         self._callback = callback
         self._timeout_ms = Duration(milliseconds=timeout_ms)
@@ -110,7 +112,12 @@ class BackgroundQueue(Generic[T]):
                 while self._queue:
                     item = self._queue.popleft()
                     try:
-                        await self._callback(item)
+                        with Measure(
+                            self._clock,
+                            name=f"rei_bgq_{self._name}",
+                            server_name=self._hs.hostname,
+                        ):
+                            await self._callback(item)
                     except defer.CancelledError:
                         raise
                     except Exception: