diff --git a/djblets/avatars/tests.py b/djblets/avatars/tests.py
index 1f04a17da2c78830c30eb1b9172ffb4861860882..bc26a09872fc5a02df7dcedb56e1e68edf1afa8d 100644
--- a/djblets/avatars/tests.py
+++ b/djblets/avatars/tests.py
@@ -696,11 +696,7 @@ class AvatarServiceRegistryTests(SpyAgency, TestCase):
         class TestRegistry(AvatarServiceRegistry):
             settings_manager_class = DummySettingsManager
 
-            def populate(self):
-                if self.populated:
-                    return
-
-                super(TestRegistry, self).populate()
+            def on_populated(self) -> None:
                 self.set_enabled_services([DummyAvatarService])
 
             def get_defaults(self):
diff --git a/djblets/registries/importer.py b/djblets/registries/importer.py
index 18e2efdc240c5c278621b504a49ca3bcbd26f99c..191b5a42c5ab28bfcf51c30bcbd6717493ca5d73 100644
--- a/djblets/registries/importer.py
+++ b/djblets/registries/importer.py
@@ -1,10 +1,20 @@
 """Import utilities for registries."""
 
+from __future__ import annotations
+
 from importlib import import_module
+from threading import Lock
 
 from django.utils.functional import SimpleLazyObject
 
 
+#: A lock used to control the importing and creation of registries.
+#:
+#: Version Added:
+#:     5.0
+_import_lock = Lock()
+
+
 def lazy_import_registry(module_path, registry_name, **kwargs):
     """Lazily import and construct a registry on access.
 
@@ -16,6 +26,14 @@ def lazy_import_registry(module_path, registry_name, **kwargs):
     This can also speed up the startup process, depending on the complexity
     of registries.
 
+    Importing and creation of registries is thread-safe. If two threads try
+    to access a registry for the first time at the same time, they'll both
+    receive the same instance.
+
+    Version Changed:
+        5.0:
+        This is now thread-safe.
+
     Args:
         module_path (str):
             The import path of the module containing the registry.
@@ -31,8 +49,9 @@ def lazy_import_registry(module_path, registry_name, **kwargs):
         A wrapper that will dynamically load and forward on to the registry.
     """
     def _create_registry():
-        mod = import_module(module_path)
+        with _import_lock:
+            mod = import_module(module_path)
 
-        return getattr(mod, registry_name)(**kwargs)
+            return getattr(mod, registry_name)(**kwargs)
 
     return SimpleLazyObject(_create_registry)
diff --git a/djblets/registries/registry.py b/djblets/registries/registry.py
index 57543d2edb257db7d66bf5ee95273669192ac9b4..3517de4a6d3013b3fcaa6dc012707f38ab5f858e 100644
--- a/djblets/registries/registry.py
+++ b/djblets/registries/registry.py
@@ -9,13 +9,16 @@ For information on writing registries, see
 from __future__ import annotations
 
 import logging
+from enum import Enum
+from threading import RLock
 from typing import (Dict, Generic, Iterable, Iterator, List, Optional,
-                    Sequence, Set, TYPE_CHECKING, Type, TypeVar)
+                    Sequence, Set, Type, TypeVar)
 
 from django.utils.translation import gettext_lazy as _
 from importlib_metadata import EntryPoint, entry_points
 from typing_extensions import Final, TypeAlias
 
+from djblets.deprecation import RemovedInDjblets70Warning
 from djblets.registries.errors import (AlreadyRegisteredError,
                                        ItemLookupError,
                                        RegistrationError)
@@ -95,6 +98,23 @@ DEFAULT_ERRORS: Final[RegistryErrorsDict] = {
 }
 
 
+class RegistryState(Enum):
+    """The operations state of a registry.
+
+    Version Added:
+        5.0
+    """
+
+    #: The registry is pending setup.
+    PENDING = 0
+
+    #: The registry is in the process of populating default items.
+    POPULATING = 1
+
+    #: The registry is populated and ready to be used.
+    READY = 2
+
+
 class Registry(Generic[RegistryItemType]):
     """An item registry.
 
@@ -110,6 +130,23 @@ class Registry(Generic[RegistryItemType]):
         class MyRegistry(Registry[MyItemType]):
             ...
 
+    Version Changed:
+        5.0:
+        * Registries now use a reentrant lock when populating, resetting,
+          registering items, or unregistering items.
+        * Added :py:attr:`state` for determining the current registry state.
+        * Deprecated :py:attr:`populated`.
+        * Added new hooks for customizing registry behavior with thread-safe
+          guarantees:
+          * :py:meth:`on_item_registering`,
+          * :py:meth:`on_item_registered`,
+          * :py:meth:`on_item_unregistering`
+          * :py:meth:`on_item_unregistered`,
+          * :py:meth:`on_populating`
+          * :py:meth:`on_populated`,
+          * :py:meth:`on_resetting`
+          * :py:meth:`on_reset`.
+
     Version Changed:
         3.1:
         Added support for specifying a registry item type when subclassing
@@ -166,24 +203,65 @@ class Registry(Generic[RegistryItemType]):
     #:     type
     lookup_error_class: Type[ItemLookupError] = ItemLookupError
 
+    ######################
+    # Instance variables #
+    ######################
+
+    #: The current state of the registry.
+    #:
+    #: Version Added:
+    #:     5.0
+    state: RegistryState
+
+    #: A set of the items stored in the registry.
+    _items: set[RegistryItemType]
+
+    #: A lock used to ensure population only happens once.
+    #:
+    #: Version Added:
+    #:     5.0
+    _lock: RLock
+
+    #: The registry of stored items.
+    #:
+    #: This is a mapping of lookup attribute names to value-to-item mappings.
+    _registry: dict[str, dict[object, RegistryItemType]]
+
     def __init__(self) -> None:
         """Initialize the registry."""
-        self._registry: Dict[str, Dict[object, RegistryItemType]] = {
+        self.state = RegistryState.PENDING
+
+        self._registry = {
             _attr_name: {}
             for _attr_name in self.lookup_attrs
         }
-        self._populated: bool = False
-        self._items: Set[RegistryItemType] = set()
+        self._lock = RLock()
+        self._items = set()
 
     @property
     def populated(self) -> bool:
         """Whether or not the registry is populated.
 
+        This can be used to determine if the registry is populated (or in
+        the process of being populated).
+
+        Consumers should check :py:attr:`state` instead, for more precise
+        tracking. This method is deprecated.
+
+        Deprecated:
+            5.0:
+            This has been replaced by :py:attr:`state` and will be removed in
+            Djblets 7.
+
         Returns:
             bool:
             Whether or not the registry is populated.
         """
-        return self._populated
+        RemovedInDjblets70Warning.warn(
+            'Registry.populated is deprecated and will be removed in '
+            'Djblets 7. Please check Registry.state instead.')
+
+        return self.state != RegistryState.PENDING
 
     def format_error(
         self,
@@ -304,38 +382,42 @@ class Registry(Generic[RegistryItemType]):
         self.populate()
         attr_values: Dict[str, object] = {}
 
-        if item in self._items:
-            raise self.already_registered_error_class(self.format_error(
-                ALREADY_REGISTERED,
-                item=item))
+        with self._lock:
+            if item in self._items:
+                raise self.already_registered_error_class(self.format_error(
+                    ALREADY_REGISTERED,
+                    item=item))
 
-        registry_map = self._registry
+            self.on_item_registering(item)
 
-        for attr_name in self.lookup_attrs:
-            attr_map = registry_map[attr_name]
+            registry_map = self._registry
 
-            try:
-                attr_value = getattr(item, attr_name)
+            for attr_name in self.lookup_attrs:
+                attr_map = registry_map[attr_name]
 
-                if attr_value in attr_map:
-                    raise self.already_registered_error_class(
-                        self.format_error(ATTRIBUTE_REGISTERED,
-                                          item=item,
-                                          duplicate=attr_map[attr_value],
-                                          attr_name=attr_name,
-                                          attr_value=attr_value))
-
-                attr_values[attr_name] = attr_value
-            except AttributeError:
-                raise RegistrationError(self.format_error(
-                    MISSING_ATTRIBUTE,
-                    item=item,
-                    attr_name=attr_name))
+                try:
+                    attr_value = getattr(item, attr_name)
+
+                    if attr_value in attr_map:
+                        raise self.already_registered_error_class(
+                            self.format_error(ATTRIBUTE_REGISTERED,
+                                              item=item,
+                                              duplicate=attr_map[attr_value],
+                                              attr_name=attr_name,
+                                              attr_value=attr_value))
 
-        for attr_name, attr_value in attr_values.items():
-            registry_map[attr_name][attr_value] = item
+                    attr_values[attr_name] = attr_value
+                except AttributeError:
+                    raise RegistrationError(self.format_error(
+                        MISSING_ATTRIBUTE,
+                        item=item,
+                        attr_name=attr_name))
 
-        self._items.add(item)
+            for attr_name, attr_value in attr_values.items():
+                registry_map[attr_name][attr_value] = item
+
+            self._items.add(item)
+            self.on_item_registered(item)
 
     def unregister_by_attr(
         self,
@@ -357,18 +439,23 @@ class Registry(Generic[RegistryItemType]):
         """
         self.populate()
 
-        try:
-            attr_map = self._registry[attr_name]
-        except KeyError:
-            raise self.lookup_error_class(self.format_error(
-                INVALID_ATTRIBUTE, attr_name=attr_name))
-        try:
-            item = attr_map[attr_value]
-        except KeyError:
-            raise self.lookup_error_class(self.format_error(
-                NOT_REGISTERED, attr_name=attr_name, attr_value=attr_value))
+        with self._lock:
+            try:
+                attr_map = self._registry[attr_name]
+            except KeyError:
+                raise self.lookup_error_class(self.format_error(
+                    INVALID_ATTRIBUTE,
+                    attr_name=attr_name))
+
+            try:
+                item = attr_map[attr_value]
+            except KeyError:
+                raise self.lookup_error_class(self.format_error(
+                    NOT_REGISTERED,
+                    attr_name=attr_name,
+                    attr_value=attr_value))
 
-        self.unregister(item)
+            self.unregister(item)
 
     def unregister(
         self,
@@ -386,33 +473,49 @@ class Registry(Generic[RegistryItemType]):
         """
         self.populate()
 
-        try:
-            self._items.remove(item)
-        except KeyError:
-            raise self.lookup_error_class(self.format_error(UNREGISTER,
-                                                            item=item))
+        with self._lock:
+            self.on_item_unregistering(item)
 
-        registry_map = self._registry
+            try:
+                self._items.remove(item)
+            except KeyError:
+                raise self.lookup_error_class(self.format_error(UNREGISTER,
+                                                                item=item))
+
+            registry_map = self._registry
 
-        for attr_name in self.lookup_attrs:
-            attr_value = getattr(item, attr_name)
-            del registry_map[attr_name][attr_value]
+            for attr_name in self.lookup_attrs:
+                attr_value = getattr(item, attr_name)
+                del registry_map[attr_name][attr_value]
+
+            self.on_item_unregistered(item)
 
     def populate(self) -> None:
         """Ensure the registry is populated.
 
         Calling this method when the registry is populated will have no effect.
         """
-        if self._populated:
+        if self.state == RegistryState.READY:
             return
 
-        self._populated = True
+        with self._lock:
+            if self.state != RegistryState.PENDING:
+                # This thread is actively populating the registry, or has been
+                # populated while waiting for the lock to be released. We can
+                # bail here.
+                return
+
+            self.state = RegistryState.POPULATING
+            self.on_populating()
 
-        for item in self.get_defaults():
-            self.register(item)
+            for item in self.get_defaults():
+                self.register(item)
 
-        registry_populating.send(sender=type(self),
-                                 registry=self)
+            self.on_populated()
+            self.state = RegistryState.READY
+
+            registry_populating.send(sender=type(self),
+                                     registry=self)
 
     def get_defaults(self) -> Iterable[RegistryItemType]:
         """Return the default items for the registry.
@@ -431,13 +534,160 @@ class Registry(Generic[RegistryItemType]):
         This will result in the registry containing no entries. Any call to a
         method that would populate the registry will repopulate it.
         """
-        if self._populated:
-            for item in self._items.copy():
-                self.unregister(item)
+        with self._lock:
+            if self.state == RegistryState.READY:
+                self.on_resetting()
+
+                for item in self._items.copy():
+                    self.unregister(item)
+
+                assert len(self._items) == 0
+
+                self.on_reset()
+                self.state = RegistryState.PENDING
+
+    def on_item_registering(
+        self,
+        item: RegistryItemType,
+        /,
+    ) -> None:
+        """Handle extra steps before registering an item.
+
+        This can be used by subclasses to perform preparation steps before
+        registering an item. It's run before the item is validated and then
+        registered.
+
+        Validation can be performed in this method.
+
+        The method is thread-safe.
+
+        Version Added:
+            5.0
+
+        Args:
+            item (object):
+                The item to register.
+
+        Raises:
+            djblets.registries.errors.RegistrationError:
+                There's an error registering this item.
+        """
+        pass
+
+    def on_item_registered(
+        self,
+        item: RegistryItemType,
+        /,
+    ) -> None:
+        """Handle extra steps after registering an item.
+
+        This can be used by subclasses to perform additional steps when an
+        item is registered. It's run after the main registration occurs.
+
+        The method is thread-safe.
+
+        Version Added:
+            5.0
+
+        Args:
+            item (object):
+                The item that was registered.
+        """
+        pass
+
+    def on_item_unregistering(
+        self,
+        item: RegistryItemType,
+        /,
+    ) -> None:
+        """Handle extra steps before unregistering an item.
+
+        This can be used by subclasses to perform additional steps before
+        validating and unregistering an item.
+
+        The method is thread-safe.
+
+        Version Added:
+            5.0
+
+        Args:
+            item (object):
+                The item to unregister.
+        """
+        pass
+
+    def on_item_unregistered(
+        self,
+        item: RegistryItemType,
+        /,
+    ) -> None:
+        """Handle extra steps after unregistering an item.
+
+        This can be used by subclasses to perform additional steps when an
+        item is unregistered. It's run after the main unregistration occurs.
+
+        The method is thread-safe.
+
+        Version Added:
+            5.0
+
+        Args:
+            item (object):
+                The item that was unregistered.
+        """
+        pass
+
+    def on_populating(self) -> None:
+        """Handle extra steps before a registry is populated.
+
+        This can be used by subclasses to perform additional steps before the
+        registry is populated.
+
+        The method is thread-safe.
+
+        Version Added:
+            5.0
+        """
+        pass
+
+    def on_populated(self) -> None:
+        """Handle extra steps after a registry is populated.
+
+        This can be used by subclasses to perform additional steps after the
+        registry is populated. It's run after the main population occurs.
+
+        The method is thread-safe.
+
+        Version Added:
+            5.0
+        """
+        pass
+
+    def on_resetting(self) -> None:
+        """Handle extra steps before resetting the registry.
+
+        This can be used by subclasses to perform additional steps before the
+        registry is reset. It's run before the main reset operations occur.
+
+        The method is thread-safe.
+
+        Version Added:
+            5.0
+        """
+        pass
 
-            self._populated = False
+    def on_reset(self) -> None:
+        """Handle extra steps after a registry is reset.
 
-        assert len(self._items) == 0
+        This can be used by subclasses to perform additional steps after the
+        registry is reset. It's run after the main reset operations occur.
+
+        The method is thread-safe.
+
+        Version Added:
+            5.0
+        """
+        pass
 
     def __iter__(self) -> Iterator[RegistryItemType]:
         """Iterate through all items in the registry.
@@ -537,45 +787,46 @@ class OrderedRegistry(Registry[RegistryItemType]):
         self._by_id: Dict[int, RegistryItemType] = {}
         self._key_order: List[int] = []
 
-    def register(
+    def on_item_registered(
         self,
         item: RegistryItemType,
+        /,
     ) -> None:
-        """Register an item.
+        """Handle extra steps before registering an item.
 
-        Args:
-            item (object):
-                The item to register with the class.
+        This will place the item in sequential order.
 
-        Raises:
-            djblets.registries.errors.RegistrationError:
-                Raised if the item is missing one of the required attributes.
+        Subclasses that override this to perform additional post-registration
+        operations must first call this method.
 
-            djblets.registries.errors.AlreadyRegisteredError:
-                Raised if the item is already registered or if the item shares
-                an attribute name, attribute value pair with another item in
-                the registry.
+        Version Added:
+            5.0
+
+        Args:
+            item (object):
+                The item that was registered.
         """
-        super(OrderedRegistry, self).register(item)
         item_id = id(item)
         self._key_order.append(item_id)
         self._by_id[item_id] = item
 
-    def unregister(
+    def on_item_unregistered(
         self,
         item: RegistryItemType,
+        /,
     ) -> None:
-        """Unregister an item from the registry.
+        """Handle extra steps after unregistering an item.
+
+        Subclasses that override this to perform additional
+        post-unregistration operations must first call this method.
+
+        Version Added:
+            5.0
 
         Args:
             item (object):
-                The item to unregister. This must be present in the registry.
-
-        Raises:
-            djblets.registries.errors.ItemLookupError:
-                Raised if the item is not found in the registry.
+                The item that was unregistered.
         """
-        super(OrderedRegistry, self).unregister(item)
         item_id = id(item)
         del self._by_id[item_id]
         self._key_order.remove(item_id)
diff --git a/djblets/registries/tests.py b/djblets/registries/tests.py
index aff01b1992fecf237665e6d337bc53139424e21b..accfcbf8f1d376713650a9670181edb1a6bc8454 100644
--- a/djblets/registries/tests.py
+++ b/djblets/registries/tests.py
@@ -1,4 +1,10 @@
-from typing import Any, Iterable
+"""Unit tests for djblets.registries."""
+
+from __future__ import annotations
+
+import time
+from threading import Lock, Thread
+from typing import Any, Callable, Iterable
 
 from kgb import SpyAgency
 
@@ -56,6 +62,47 @@ class Item:
                     for attr_name in self._attrs))
 
 
+class ThreadRegistry(Registry[Item]):
+    """A registry used for thread tests.
+
+    Version Added:
+        5.0
+    """
+
+    lookup_attrs = ('id',)
+
+    ######################
+    # Instance variables #
+    ######################
+
+    count: int
+    sleep_lock: Lock
+
+    def __init__(self, *args, **kwargs) -> None:
+        """Initialize the registry.
+
+        Args:
+            *args (tuple):
+                Positional arguments to pass to the parent.
+
+            **kwargs (tuple):
+                Keyword arguments to pass to the parent.
+        """
+        super().__init__(*args, **kwargs)
+
+        self.count = 0
+        self.sleep_lock = Lock()
+
+    def sleep_first_thread(self) -> None:
+        """Sleep the first time a thread calls this method."""
+        with self.sleep_lock:
+            do_sleep = (self.count == 0)
+            self.count += 1
+
+        if do_sleep:
+            time.sleep(0.1)
+
+
 class RegistryTests(SpyAgency, TestCase):
     """Tests for djblets.registries.Registry."""
 
@@ -73,9 +120,13 @@ class RegistryTests(SpyAgency, TestCase):
         with self.assertRaises(ItemLookupError):
             r.get('foo', 'bar')
 
-    def test_register_item(self):
+    def test_register_item(self) -> None:
         """Testing Registry.register_item"""
         r = Registry()
+
+        self.spy_on(r.on_item_registering)
+        self.spy_on(r.on_item_registered)
+
         items = [1, 2, 3]
 
         for item in items:
@@ -83,11 +134,69 @@ class RegistryTests(SpyAgency, TestCase):
 
         self.assertEqual(set(r), set(items))
 
+        self.assertSpyCallCount(r.on_item_registering, 3)
+        self.assertSpyCalledWith(r.on_item_registering, 1)
+        self.assertSpyCalledWith(r.on_item_registering, 2)
+        self.assertSpyCalledWith(r.on_item_registering, 3)
+
+        self.assertSpyCallCount(r.on_item_registered, 3)
+        self.assertSpyCalledWith(r.on_item_registered, 1)
+        self.assertSpyCalledWith(r.on_item_registered, 2)
+        self.assertSpyCalledWith(r.on_item_registered, 3)
+
+    def test_register_item_with_thread_conflict(self) -> None:
+        """Testing Registry.register_item with same item in different threads
+        """
+        def _thread_main(
+            secs: float,
+            item: Item,
+            expect_fail: bool,
+        ) -> None:
+            time.sleep(secs)
+
+            if expect_fail:
+                with self.assertRaises(AlreadyRegisteredError):
+                    registry.register(item)
+            else:
+                registry.register(item)
+
+        class TestRegistry(ThreadRegistry):
+            def on_item_registering(
+                self,
+                item: Item,
+            ) -> None:
+                self.sleep_first_thread()
+
+        registry = TestRegistry()
+        registry.populate()
+        self.spy_on(registry.on_item_registering)
+        self.spy_on(registry.on_item_registered)
+
+        self.assertEqual(len(registry), 0)
+
+        item1 = Item(id=0)
+        item2 = Item(id=0)
+
+        self._run_threads(
+            target=_thread_main,
+            threads_args=[
+                [0.1, item1, True],
+                [0.05, item2, False],
+            ])
+
+        self.assertIs(registry.get('id', 0), item2)
+
+        self.assertSpyCallCount(registry.on_item_registering, 2)
+        self.assertSpyCallCount(registry.on_item_registered, 1)
+
     def test_unregister_item(self):
         """Testing Registry.unregister_item"""
         r = Registry()
         items = [1, 2, 3]
 
+        self.spy_on(r.on_item_unregistering)
+        self.spy_on(r.on_item_unregistered)
+
         for item in items:
             r.register(item)
 
@@ -96,6 +205,59 @@ class RegistryTests(SpyAgency, TestCase):
 
         self.assertEqual(set(r), set())
 
+        self.assertSpyCallCount(r.on_item_unregistering, 3)
+        self.assertSpyCalledWith(r.on_item_unregistering, 1)
+        self.assertSpyCalledWith(r.on_item_unregistering, 2)
+        self.assertSpyCalledWith(r.on_item_unregistering, 3)
+
+        self.assertSpyCallCount(r.on_item_unregistered, 3)
+        self.assertSpyCalledWith(r.on_item_unregistered, 1)
+        self.assertSpyCalledWith(r.on_item_unregistered, 2)
+        self.assertSpyCalledWith(r.on_item_unregistered, 3)
+
+    def test_unregister_item_with_thread_conflict(self) -> None:
+        """Testing Registry.unregister_item with same item in different
+        threads
+        """
+        def _thread_main(
+            secs: float,
+            expect_fail: bool,
+        ) -> None:
+            time.sleep(secs)
+
+            if expect_fail:
+                with self.assertRaises(ItemLookupError):
+                    registry.unregister(item1)
+            else:
+                registry.unregister(item1)
+
+        class TestRegistry(ThreadRegistry):
+            def on_item_unregistering(
+                self,
+                item: Item,
+            ) -> None:
+                self.sleep_first_thread()
+
+        registry = TestRegistry()
+        self.spy_on(registry.on_item_unregistering)
+        self.spy_on(registry.on_item_unregistered)
+
+        item1 = Item(id=0)
+        registry.register(item1)
+
+        self.assertEqual(len(registry), 1)
+
+        self._run_threads(
+            target=_thread_main,
+            threads_args=[
+                [0.1, True],
+                [0.05, False],
+            ])
+
+        self.assertEqual(len(registry), 0)
+        self.assertSpyCallCount(registry.on_item_unregistering, 2)
+        self.assertSpyCallCount(registry.on_item_unregistered, 1)
+
     def test_unregister_removes_attr_lookups(self):
         """Testing Registry.unregister removes lookup entries"""
         class TestRegistry(Registry[Item]):
@@ -115,6 +277,80 @@ class RegistryTests(SpyAgency, TestCase):
 
         self.assertEqual(len(r), 0)
 
+    def test_populate(self) -> None:
+        """Testing Registry.populate"""
+        item1 = Item(id=0)
+        item2 = Item(id=1)
+
+        class TestRegistry(Registry[Item]):
+            lookup_attrs = ('id',)
+
+            def get_defaults(self):
+                yield item1
+                yield item2
+
+        self.spy_on(registry_populating.send)
+
+        r = TestRegistry()
+        self.spy_on(r.on_populated)
+        self.spy_on(r.on_populating)
+
+        r.populate()
+
+        self.assertEqual(r._registry, {
+            'id': {
+                0: item1,
+                1: item2,
+            },
+        })
+
+        self.assertSpyCalled(r.on_populated)
+        self.assertSpyCalled(r.on_populating)
+
+        self.assertIs(r.get('id', item1.id), item1)
+        self.assertIs(r.get('id', item2.id), item2)
+
+        self.assertSpyCalledWith(
+            registry_populating.send,
+            sender=TestRegistry,
+            registry=r)
+
+    def test_populate_with_thread_conflict(self) -> None:
+        """Testing Registry.populate with multiple threads"""
+        def _thread_main(
+            secs: float,
+        ) -> None:
+            time.sleep(secs)
+            registry.populate()
+
+            self.assertEqual(list(registry._items), [item1, item2])
+
+        class TestRegistry(ThreadRegistry):
+            def get_defaults(self):
+                yield item1
+                yield item2
+
+            def on_populating(self) -> None:
+                self.sleep_first_thread()
+
+        registry = TestRegistry()
+        self.spy_on(registry.on_populating)
+        self.spy_on(registry.on_populated)
+
+        item1 = Item(id=0)
+        item2 = Item(id=1)
+
+        self._run_threads(
+            target=_thread_main,
+            threads_args=[
+                [0.1],
+                [0.05],
+            ])
+
+        self.assertEqual(list(registry._items), [item1, item2])
+        self.assertSpyCallCount(registry.on_populating, 1)
+        self.assertSpyCallCount(registry.on_populated, 1)
+
     def test_population_on_register(self):
         """Testing Registry.register_item triggers population before
         registration
@@ -198,6 +434,73 @@ class RegistryTests(SpyAgency, TestCase):
         with self.assertRaises(RegistrationError):
             r.register(Item())
 
+    def test_reset(self) -> None:
+        """Testing Registry.reset"""
+        item1 = Item(id=0)
+        item2 = Item(id=1)
+
+        class TestRegistry(Registry[Item]):
+            lookup_attrs = ('id',)
+
+            def get_defaults(self):
+                yield item1
+                yield item2
+
+        self.spy_on(registry_populating.send)
+
+        r = TestRegistry()
+        self.spy_on(r.on_resetting)
+        self.spy_on(r.on_reset)
+
+        r.populate()
+        r.reset()
+
+        self.assertEqual(r._registry, {
+            'id': {},
+        })
+
+        self.assertSpyCalled(r.on_resetting)
+        self.assertSpyCalled(r.on_reset)
+
+    def test_reset_with_thread_conflict(self) -> None:
+        """Testing Registry.reset with multiple threads"""
+        def _thread_main(
+            secs: float,
+        ) -> None:
+            self.assertEqual(list(registry._items), [item1, item2])
+
+            time.sleep(secs)
+            registry.reset()
+
+            self.assertEqual(list(registry._items), [])
+
+        class TestRegistry(ThreadRegistry):
+            def get_defaults(self):
+                yield item1
+                yield item2
+
+            def on_resetting(self) -> None:
+                self.sleep_first_thread()
+
+        item1 = Item(id=0)
+        item2 = Item(id=1)
+
+        registry = TestRegistry()
+        registry.populate()
+        self.spy_on(registry.on_resetting)
+        self.spy_on(registry.on_reset)
+
+        self._run_threads(
+            target=_thread_main,
+            threads_args=[
+                [0.1],
+                [0.05],
+            ])
+
+        self.assertEqual(len(registry._items), 0)
+        self.assertSpyCallCount(registry.on_resetting, 1)
+        self.assertSpyCallCount(registry.on_reset, 1)
+
     def test_contains(self):
         """Testing Registry.__contains__"""
         r = Registry()
@@ -219,6 +522,42 @@ class RegistryTests(SpyAgency, TestCase):
                                       'The foo "1" is unregistered.'):
             r.unregister(1)
 
+    def _run_threads(
+        self,
+        target: Callable[..., None],
+        threads_args: list[list[Any]],
+    ) -> None:
+        """Run tests with a specified number of threads.
+
+        This will create multiple threads with the same target and different
+        arguments, starting them up and then joining them back to the main
+        thread.
+
+        Version Added:
+            5.0
+
+        Args:
+            target (callable):
+                The function for each thread to call when started.
+
+            threads_args (list):
+                Arguments to provide for each created thread.
+
+                The number of items in this list dictates the number of
+                threads.
+        """
+        threads = [
+            Thread(target=target,
+                   args=thread_args)
+            for thread_args in threads_args
+        ]
+
+        for thread in threads:
+            thread.start()
+
+        for thread in threads:
+            thread.join()
+
 
 class OrderedRegistryTests(TestCase):
     """Tests for djblets.registries.registry.OrderedRegistry."""