[gym.vector]: a generic`close` with user-defined `close_extras`
diff --git a/gym/vector/async_vector_env.py b/gym/vector/async_vector_env.py
index d1d4a0809a2..3ef90422863 100644
--- a/gym/vector/async_vector_env.py
+++ b/gym/vector/async_vector_env.py
@@ -241,7 +241,7 @@ def step_wait(self, timeout=None):
         return (deepcopy(self.observations) if self.copy else self.observations,
                 np.array(rewards), np.array(dones, dtype=np.bool_), infos)
 
-    def close(self, timeout=None, terminate=False):
+    def close_extras(self, timeout=None, terminate=False):
         """
         Parameters
         ----------
@@ -254,12 +254,6 @@ def close(self, timeout=None, terminate=False):
             If `True`, then the `close` operation is forced and all processes
             are terminated.
         """
-        if self.closed:
-            return
-
-        if self.viewer is not None:
-            self.viewer.close()
-
         timeout = 0 if terminate else timeout
         try:
             if self._state != AsyncState.DEFAULT:
@@ -288,8 +282,6 @@ def close(self, timeout=None, terminate=False):
         for process in self.processes:
             process.join()
 
-        self.closed = True
-
     def _poll(self, timeout=None):
         self._assert_is_running()
         if timeout is None:
@@ -338,11 +330,6 @@ def _raise_if_errors(self, successes):
         logger.error('Raising the last exception back to the main process.')
         raise exctype(value)
 
-    def __del__(self):
-        if hasattr(self, 'closed'):
-            if not self.closed:
-                self.close(terminate=True)
-
 
 def _worker(index, env_fn, pipe, parent_pipe, shared_memory, error_queue):
     assert shared_memory is None
diff --git a/gym/vector/sync_vector_env.py b/gym/vector/sync_vector_env.py
index 379977ae9e7..b3d4eb03484 100644
--- a/gym/vector/sync_vector_env.py
+++ b/gym/vector/sync_vector_env.py
@@ -83,16 +83,8 @@ def step_wait(self):
         return (deepcopy(self.observations) if self.copy else self.observations,
             np.copy(self._rewards), np.copy(self._dones), infos)
 
-    def close(self):
-        if self.closed:
-            return
-        if self.viewer is not None:
-            self.viewer.close()
-
-        for env in self.envs:
-            env.close()
-
-        self.closed = True
+    def close_extras(self, **kwargs):
+        [env.close() for env in self.envs]
 
     def _check_observation_spaces(self):
         for env in self.envs:
diff --git a/gym/vector/vector_env.py b/gym/vector/vector_env.py
index aabe0f1b0d5..973fff803a2 100644
--- a/gym/vector/vector_env.py
+++ b/gym/vector/vector_env.py
@@ -91,6 +91,34 @@ def step(self, actions):
         self.step_async(actions)
         return self.step_wait()
 
+    def close_extras(self, **kwargs):
+        r"""Clean up the extra resources e.g. beyond what's in this base class. """
+        raise NotImplementedError()
+
+    def close(self, **kwargs):
+        r"""Close all sub-environments and release resources.
+        
+        It also closes all the existing image viewers, then calls :meth:`close_extras` and set
+        :attr:`closed` as ``True``. 
+        
+        .. warning::
+        
+            This function itself does not close the environments, it should be handled
+            in :meth:`close_extras`. This is generic for both synchronous and asynchronous 
+            vectorized environments. 
+        
+        .. note::
+        
+            This will be automatically called when garbage collected or program exited. 
+            
+        """
+        if self.closed:
+            return
+        if self.viewer is not None:
+            self.viewer.close()
+        self.close_extras(**kwargs)
+        self.closed = True
+
     def seed(self, seeds=None):
         """
         Parameters
@@ -104,8 +132,7 @@ def seed(self, seeds=None):
         """
         pass
 
-
     def __del__(self):
         if hasattr(self, 'closed'):
             if not self.closed:
-                self.close()
+                self.close(terminate=True)
