diff --git a/jsonschema/_utils.py b/jsonschema/_utils.py
index 7b1c33a0a..d3ae67eef 100644
--- a/jsonschema/_utils.py
+++ b/jsonschema/_utils.py
@@ -215,3 +215,42 @@ def uniq(container):
                     return False
                 seen.append(e)
     return True
+
+
+def subschemas(schema, key=u'id'):
+    """
+        Finds all explicitly identified (sub)schemas contained in the provided ``schema``, including the root schema.
+        The optional ``key`` parameter is to allow for easier interop between draft-03/04 and draft-06+.
+
+        Arguments:
+
+            schema (dict):
+
+                The schema to extract all constituent (sub)schemas from.
+
+            key (str):
+
+                The keyword used to define a URI reference for a schema. id for draft-03/04 and $id for draft-06+.
+
+        Returns:
+
+            list: A list of the (sub)schemas defined within the provided ``schema``.
+
+        """
+    schemas = list()
+    if not isinstance(schema, dict):
+        return schemas
+    schema_id = schema.get(key, None)
+    if schema_id is not None and isinstance(schema_id, str_types):
+        schemas.append(schema)
+    for key, value in schema.items():
+        # Recurse if the value is a dictionary.
+        if isinstance(value, dict):
+            schemas.extend(subschemas(value))
+        # Check all list members. We don't have to worry about iterables due to raw schema type mapping.
+        if isinstance(value, list):
+            for item in value:
+                # Recurse if the value is a dictionary.
+                if isinstance(item, dict):
+                    schemas.extend(subschemas(item))
+    return schemas
diff --git a/jsonschema/tests/test_validators.py b/jsonschema/tests/test_validators.py
index 2df497148..b2521f4c5 100644
--- a/jsonschema/tests/test_validators.py
+++ b/jsonschema/tests/test_validators.py
@@ -897,7 +897,8 @@ def test_custom_uri_scheme_handlers(self):
 
     def test_cache_remote_on(self):
         ref = "foo://bar"
-        foo_handler = mock.Mock()
+        foo_handler = mock.MagicMock(spec=dict())
+        foo_handler.__getitem__.side_effect = ''
         resolver = RefResolver(
             "", {}, cache_remote=True, handlers={"foo": foo_handler},
         )
@@ -909,7 +910,8 @@ def test_cache_remote_on(self):
 
     def test_cache_remote_off(self):
         ref = "foo://bar"
-        foo_handler = mock.Mock()
+        foo_handler = mock.MagicMock(spec=dict())
+        foo_handler.__getitem__.side_effect = ''
         resolver = RefResolver(
             "", {}, cache_remote=False, handlers={"foo": foo_handler},
         )
diff --git a/jsonschema/validators.py b/jsonschema/validators.py
index c8148fe3f..95ed0e65e 100644
--- a/jsonschema/validators.py
+++ b/jsonschema/validators.py
@@ -299,6 +299,7 @@ def __init__(
         )
         self.store.update(store)
         self.store[base_uri] = referrer
+        self.subschema_store = self.extract_store_subschemas()
 
         self._urljoin_cache = urljoin_cache
         self._remote_cache = remote_cache
@@ -322,6 +323,14 @@ def from_schema(cls, schema, *args, **kwargs):
 
         return cls(schema.get(u"id", u""), schema, *args, **kwargs)
 
+    def extract_store_subschemas(self):
+        subschemas = dict()
+        for base_id, schema in self.store.items():
+            subschemas[base_id] = {
+                subschema[u"id"].lstrip(u'#'): subschema for subschema in _utils.subschemas(schema)
+            }
+        return subschemas
+
     def push_scope(self, scope):
         self._scopes_stack.append(
             self._urljoin_cache(self.resolution_scope, scope),
@@ -407,6 +416,12 @@ def resolve_fragment(self, document, fragment):
 
         """
 
+        # First try looking up the fragment schema reference in the subschema_store.
+        try:
+            return self.subschema_store[document[u'id']][fragment]
+        except (KeyError, LookupError, TypeError):
+            pass
+
         fragment = fragment.lstrip(u"/")
         parts = unquote(fragment).split(u"/") if fragment else []
 
@@ -480,6 +495,9 @@ def resolve_remote(self, uri):
 
         if self.cache_remote:
             self.store[uri] = result
+            self.subschema_store[uri] = {
+                subschema[u"id"].lstrip(u'#'): subschema for subschema in _utils.subschemas(result)
+            }
         return result