diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index aabb5c1..caa614c 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -1,6 +1,7 @@
 Next release
 ------------
 - Added ``Connector.timer`` method (addresses :issue:`8`)
+- Implement ``Timer`` abstraction from python-statsd
 
 :tag:`0.1.0 <0.0.1...0.1.0>` (10-May-2021)
 ------------------------------------------
diff --git a/README.rst b/README.rst
index 117fb64..5584c4c 100644
--- a/README.rst
+++ b/README.rst
@@ -36,15 +36,29 @@ metric to send and the task consumes the internal queue when it is connected.
 The following convenience methods are available.  You can also call ``inject_metric`` for complete control over
 the payload.
 
-+--------------+--------------------------------------+
-| ``incr``     | Increment a counter metric           |
-+--------------+--------------------------------------+
-| ``decr``     | Decrement a counter metric           |
-+--------------+--------------------------------------+
-| ``gauge``    | Adjust or set a gauge metric         |
-+--------------+--------------------------------------+
-| ``timing``   | Append a duration to a timer metric  |
-+--------------+--------------------------------------+
++--------------+--------------------------------------------------------------+
+| ``incr``     | Increment a counter metric                                   |
++--------------+--------------------------------------------------------------+
+| ``decr``     | Decrement a counter metric                                   |
++--------------+--------------------------------------------------------------+
+| ``gauge``    | Adjust or set a gauge metric                                 |
++--------------+--------------------------------------------------------------+
+| ``timer``    | Append a duration to a timer metric using a context manager  |
++--------------+--------------------------------------------------------------+
+| ``timing``   | Append a duration to a timer metric                          |
++--------------+--------------------------------------------------------------+
+
+If you are a `python-statsd`_ user, then the method names should look very familiar.  That is quite intentional.
+I like the interface and many others do as well.  There is one very very important difference though -- the
+``timing`` method takes the duration as the number of **seconds** as a :class:`float` instead of the number of
+milliseconds.
+
+.. warning::
+
+   If you are accustomed to using `python-statsd`_, be aware that the ``timing`` method expects the number of
+   seconds as a :class:`float` instead of the number of milliseconds.
+
+.. _python-statsd: https://statsd.readthedocs.io/en/latest/
 
 Tornado helpers
 ===============
diff --git a/docs/index.rst b/docs/index.rst
index 96657cc..58e62af 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -46,6 +46,7 @@ Reference
 
 .. autoclass:: sprockets_statsd.statsd.Connector
    :members:
+   :inherited-members: incr, decr, gauge, timer, timing
 
 Tornado helpers
 ---------------
@@ -63,6 +64,9 @@ Internals
 .. autoclass:: sprockets_statsd.statsd.Processor
    :members:
 
+.. autoclass:: sprockets_statsd.statsd.Timer
+   :members:
+
 .. autoclass:: sprockets_statsd.statsd.StatsdProtocol
    :members:
 
diff --git a/sprockets_statsd/statsd.py b/sprockets_statsd/statsd.py
index d8080e7..f8bb7ce 100644
--- a/sprockets_statsd/statsd.py
+++ b/sprockets_statsd/statsd.py
@@ -1,5 +1,4 @@
 import asyncio
-import contextlib
 import logging
 import socket
 import time
@@ -47,6 +46,80 @@ class ThrottleGuard:
         self.counter = 0
 
 
+class Timer:
+    """Implement Timer interface from python-statsd.
+
+    Instances of this class are returned from
+    :meth:`.AbstractConnector.timer` to maintain some compatibility
+    with `python-statsd`_.  You should not create instances of this
+    class yourself.
+
+    This implementation follows the careful protocol created by
+    python-statsd in that it raises :exc:`RuntimeError` in the
+    following cases:
+
+    * :meth:`.stop` is called before calling :meth:`.start`
+    * :meth:`.send` is called before calling :meth:`.stop`
+      which requires a prior call to :meth:`.start`
+
+    The call to :meth:`.send` clears the timing values so calling it
+    twice in a row will result in a :exc:`RuntimeError` as well.
+
+    """
+    def __init__(self, connector: 'AbstractConnector', path: str):
+        self._connector = connector
+        self._path = path
+        self._start_time, self._finish_time = None, None
+
+    def start(self) -> 'Timer':
+        """Start the timer and return `self`."""
+        self._start_time = time.time()
+        return self
+
+    def stop(self, send: bool = True) -> 'Timer':
+        """Stop the timer and send the timing.
+
+        :param send: immediately send the recorded timing to the
+            processor
+
+        You can delay sending the timing by setting the `send`
+        parameter to :data:`False`.
+
+        """
+        if self._start_time is None:
+            raise RuntimeError('Timer.stop called before start')
+        self._finish_time = time.time()
+        if send:
+            self.send()
+        return self
+
+    def send(self):
+        """Send the recorded timing to the connector.
+
+        This method will raise a :exc:`RuntimeError` if a timing has
+        not been recorded by calling :meth:`start` and :meth:`stop` in
+        sequence.
+
+        """
+        if self._start_time is None:
+            raise RuntimeError('Timer.send called before start')
+        if self._finish_time is None:
+            raise RuntimeError('Timer.send called before stop')
+
+        self._connector.timing(
+            self._path,
+            max(self._finish_time, self._start_time) - self._start_time)
+        self._start_time, self._finish_time = None, None
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()
+        return False
+
+
 class AbstractConnector:
     """StatsD connector that does not send metrics or connect.
 
@@ -110,19 +183,13 @@ class AbstractConnector:
         """
         self.inject_metric(f'timers.{path}', str(seconds * 1000.0), 'ms')
 
-    @contextlib.contextmanager
-    def timer(self, path):
+    def timer(self, path) -> Timer:
         """Send a timer metric using a context manager.
 
         :param path: timer to append the measured time to
 
         """
-        start = time.time()
-        try:
-            yield
-        finally:
-            fini = time.time()
-            self.timing(path, max(fini, start) - start)
+        return Timer(self, path)
 
 
 class Connector(AbstractConnector):
diff --git a/tests/test_processor.py b/tests/test_processor.py
index e2f98d3..eede86f 100644
--- a/tests/test_processor.py
+++ b/tests/test_processor.py
@@ -460,3 +460,83 @@ class ConnectorOptionTests(ProcessorTestCase):
         for _ in range(connector.processor.queue.qsize()):
             metric = await connector.processor.queue.get()
             self.assertEqual(metric, b'counters.counter:1|c')
+
+
+class ConnectorTimerTests(ProcessorTestCase):
+    ip_protocol = socket.IPPROTO_TCP
+
+    async def asyncSetUp(self):
+        await super().asyncSetUp()
+        self.connector = statsd.Connector(self.statsd_server.host,
+                                          self.statsd_server.port)
+        await self.connector.start()
+        await self.wait_for(self.statsd_server.client_connected.acquire())
+
+    async def asyncTearDown(self):
+        await self.wait_for(self.connector.stop())
+        await super().asyncTearDown()
+
+    async def test_that_stop_raises_if_not_started(self):
+        timer = self.connector.timer('whatever')
+        with self.assertRaises(RuntimeError):
+            timer.stop()
+
+    async def test_that_start_returns_instance(self):
+        timer = self.connector.timer('whatever')
+        self.assertIs(timer, timer.start())
+
+    async def test_that_stop_returns_instance(self):
+        timer = self.connector.timer('whatever')
+        timer.start()
+        self.assertIs(timer, timer.stop())
+
+    async def test_that_timing_is_sent_by_stop(self):
+        timer = self.connector.timer('whatever')
+        timer.start()
+        self.assertTrue(self.statsd_server.message_received.locked(),
+                        'timing sent to server unexpectedly')
+        timer.stop()
+        await self.wait_for(self.statsd_server.message_received.acquire())
+
+    async def test_that_timing_send_can_be_delayed(self):
+        timer = self.connector.timer('whatever')
+        timer.start()
+        self.assertTrue(self.statsd_server.message_received.locked(),
+                        'timing sent to server unexpectedly')
+        timer.stop(send=False)
+        self.assertTrue(self.statsd_server.message_received.locked(),
+                        'timing sent to server unexpectedly')
+
+        timer.send()
+        await self.wait_for(self.statsd_server.message_received.acquire())
+
+    async def test_that_send_raises_when_already_sent(self):
+        timer = self.connector.timer('whatever')
+        timer.start()
+        timer.stop(send=False)
+        timer.send()
+        await self.wait_for(self.statsd_server.message_received.acquire())
+        with self.assertRaises(RuntimeError):
+            timer.send()
+
+    async def test_that_send_raises_when_not_started(self):
+        timer = self.connector.timer('whatever')
+        with self.assertRaises(RuntimeError):
+            timer.send()
+
+    async def test_that_send_raises_when_not_stopped(self):
+        timer = self.connector.timer('whatever')
+        timer.start()
+        with self.assertRaises(RuntimeError):
+            timer.send()
+
+    async def test_that_timer_can_be_reused(self):
+        timer = self.connector.timer('whatever')
+        with timer:
+            pass
+        await self.wait_for(self.statsd_server.message_received.acquire())
+        self.assertTrue(self.statsd_server.message_received.locked())
+
+        with timer:
+            pass
+        await self.wait_for(self.statsd_server.message_received.acquire())