diff --git a/changelog.d/1164.added.rst b/changelog.d/1164.added.rst
new file mode 100644
index 00000000..f6e6612b
--- /dev/null
+++ b/changelog.d/1164.added.rst
@@ -0,0 +1 @@
+Added ``loop_factory`` to pytest_asyncio.fixture and asyncio mark
diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py
index 29252b3e..0921c7d6 100644
--- a/pytest_asyncio/plugin.py
+++ b/pytest_asyncio/plugin.py
@@ -134,6 +134,7 @@ def fixture(
     *,
     scope: _ScopeName | Callable[[str, Config], _ScopeName] = ...,
     loop_scope: _ScopeName | None = ...,
+    loop_factory: Callable[[], AbstractEventLoop] | None = ...,
     params: Iterable[object] | None = ...,
     autouse: bool = ...,
     ids: (
@@ -151,6 +152,7 @@ def fixture(
     *,
     scope: _ScopeName | Callable[[str, Config], _ScopeName] = ...,
     loop_scope: _ScopeName | None = ...,
+    loop_factory: Callable[[], AbstractEventLoop] | None = ...,
     params: Iterable[object] | None = ...,
     autouse: bool = ...,
     ids: (
@@ -165,20 +167,26 @@ def fixture(
 def fixture(
     fixture_function: FixtureFunction[_P, _R] | None = None,
     loop_scope: _ScopeName | None = None,
+    loop_factory: Callable[[], AbstractEventLoop] | None = None,
     **kwargs: Any,
 ) -> (
     FixtureFunction[_P, _R]
     | Callable[[FixtureFunction[_P, _R]], FixtureFunction[_P, _R]]
 ):
     if fixture_function is not None:
-        _make_asyncio_fixture_function(fixture_function, loop_scope)
+        _make_asyncio_fixture_function(fixture_function, loop_scope, loop_factory)
         return pytest.fixture(fixture_function, **kwargs)
 
     else:
 
         @functools.wraps(fixture)
         def inner(fixture_function: FixtureFunction[_P, _R]) -> FixtureFunction[_P, _R]:
-            return fixture(fixture_function, loop_scope=loop_scope, **kwargs)
+            return fixture(
+                fixture_function,
+                loop_factory=loop_factory,
+                loop_scope=loop_scope,
+                **kwargs,
+            )
 
         return inner
 
@@ -188,12 +196,17 @@ def _is_asyncio_fixture_function(obj: Any) -> bool:
     return getattr(obj, "_force_asyncio_fixture", False)
 
 
-def _make_asyncio_fixture_function(obj: Any, loop_scope: _ScopeName | None) -> None:
+def _make_asyncio_fixture_function(
+    obj: Any,
+    loop_scope: _ScopeName | None,
+    loop_factory: Callable[[], AbstractEventLoop] | None,
+) -> None:
     if hasattr(obj, "__func__"):
         # instance method, check the function object
         obj = obj.__func__
     obj._force_asyncio_fixture = True
     obj._loop_scope = loop_scope
+    obj._loop_factory = loop_factory
 
 
 def _is_coroutine_or_asyncgen(obj: Any) -> bool:
@@ -280,7 +293,9 @@ def pytest_report_header(config: Config) -> list[str]:
 
 
 def _fixture_synchronizer(
-    fixturedef: FixtureDef, runner: Runner, request: FixtureRequest
+    fixturedef: FixtureDef,
+    runner: Runner,
+    request: FixtureRequest,
 ) -> Callable:
     """Returns a synchronous function evaluating the specified fixture."""
     fixture_function = resolve_fixture_function(fixturedef, request)
@@ -353,7 +368,7 @@ def _wrap_async_fixture(
     runner: Runner,
     request: FixtureRequest,
 ) -> Callable[AsyncFixtureParams, AsyncFixtureReturnType]:
-    @functools.wraps(fixture_function)
+    @functools.wraps(fixture_function)  # type: ignore[arg-type]
     def _async_fixture_wrapper(
         *args: AsyncFixtureParams.args,
         **kwargs: AsyncFixtureParams.kwargs,
@@ -363,8 +378,8 @@ async def setup():
             return res
 
         context = contextvars.copy_context()
-        result = runner.run(setup(), context=context)
 
+        result = runner.run(setup(), context=context)
         # Copy the context vars modified by the setup task into the current
         # context, and (if needed) add a finalizer to reset them.
         #
@@ -617,16 +632,32 @@ def pytest_pycollect_makeitem_convert_async_functions_to_subclass(
 
 
 @contextlib.contextmanager
-def _temporary_event_loop_policy(policy: AbstractEventLoopPolicy) -> Iterator[None]:
+def _temporary_event_loop_policy(
+    policy: AbstractEventLoopPolicy,
+    loop_facotry: Callable[..., AbstractEventLoop] | None,
+) -> Iterator[None]:
+
     old_loop_policy = _get_event_loop_policy()
     try:
         old_loop = _get_event_loop_no_warn()
     except RuntimeError:
         old_loop = None
+    # XXX: For some reason this function can override runner's
+    # _loop_factory (At least observed on backported versions of Runner)
+    # so we need to re-override if existing...
+    if loop_facotry:
+        _loop = loop_facotry()
+        _set_event_loop(_loop)
+    else:
+        _loop = None
+
     _set_event_loop_policy(policy)
     try:
         yield
     finally:
+        if _loop:
+            # Do not let BaseEventLoop.__del__ complain!
+            _loop.close()
         _set_event_loop_policy(old_loop_policy)
         _set_event_loop(old_loop)
 
@@ -739,10 +770,12 @@ def pytest_fixture_setup(fixturedef: FixtureDef, request) -> object | None:
         or default_loop_scope
         or fixturedef.scope
     )
+    loop_factory = getattr(fixturedef.func, "loop_factory", None)
+
     runner_fixture_id = f"_{loop_scope}_scoped_runner"
     runner = request.getfixturevalue(runner_fixture_id)
     synchronizer = _fixture_synchronizer(fixturedef, runner, request)
-    _make_asyncio_fixture_function(synchronizer, loop_scope)
+    _make_asyncio_fixture_function(synchronizer, loop_scope, loop_factory)
     with MonkeyPatch.context() as c:
         c.setattr(fixturedef, "func", synchronizer)
         hook_result = yield
@@ -765,9 +798,13 @@ def _get_marked_loop_scope(
 ) -> _ScopeName:
     assert asyncio_marker.name == "asyncio"
     if asyncio_marker.args or (
-        asyncio_marker.kwargs and set(asyncio_marker.kwargs) - {"loop_scope", "scope"}
+        asyncio_marker.kwargs
+        and set(asyncio_marker.kwargs) - {"loop_scope", "scope", "loop_factory"}
     ):
-        raise ValueError("mark.asyncio accepts only a keyword argument 'loop_scope'.")
+        raise ValueError(
+            "mark.asyncio accepts only a keyword arguments 'loop_scope' "
+            "or 'loop_factory'"
+        )
     if "scope" in asyncio_marker.kwargs:
         if "loop_scope" in asyncio_marker.kwargs:
             raise pytest.UsageError(_DUPLICATE_LOOP_SCOPE_DEFINITION_ERROR)
@@ -796,19 +833,50 @@ def _get_default_test_loop_scope(config: Config) -> Any:
 """
 
 
+def _get_loop_facotry(
+    request: FixtureRequest,
+) -> Callable[[], AbstractEventLoop] | None:
+    loop_factories = []
+    asyncio_mark = request._pyfuncitem.get_closest_marker("asyncio")
+    if asyncio_mark is not None:
+        # The loop_factory is defined on an asyncio marker
+        factory = asyncio_mark.kwargs.get("loop_factory", None)
+        loop_factories.append(factory)
+    # The loop_factory is defined in a transitive fixture
+    current_request = request
+    for r in request._iter_chain():
+        current_request = r
+        loop_factory = getattr(current_request._fixturedef.func, "_loop_factory", None)
+        loop_factories.append(loop_factory)
+    defined_loop_factories = [factory for factory in loop_factories if factory] or [
+        None
+    ]
+    print(defined_loop_factories)
+    if len(defined_loop_factories) > 1:
+        print(defined_loop_factories)
+        raise pytest.UsageError(
+            "Multiple loop factories defined for {request.scope}-scoped loop."
+        )
+    return defined_loop_factories[0]
+
+
 def _create_scoped_runner_fixture(scope: _ScopeName) -> Callable:
     @pytest.fixture(
         scope=scope,
         name=f"_{scope}_scoped_runner",
     )
     def _scoped_runner(
-        event_loop_policy,
-        request: FixtureRequest,
+        event_loop_policy: AbstractEventLoopPolicy, request: FixtureRequest
     ) -> Iterator[Runner]:
         new_loop_policy = event_loop_policy
+
+        # We need to get the factory now because
+        # _temporary_event_loop_policy can override the Runner
+        factory = _get_loop_facotry(request)
         debug_mode = _get_asyncio_debug(request.config)
-        with _temporary_event_loop_policy(new_loop_policy):
-            runner = Runner(debug=debug_mode).__enter__()
+        with _temporary_event_loop_policy(new_loop_policy, factory):
+            runner = Runner(debug=debug_mode, loop_factory=factory).__enter__()
+
             try:
                 yield runner
             except Exception as e:
diff --git a/tests/async_fixtures/test_async_fixtures_contextvars.py b/tests/async_fixtures/test_async_fixtures_contextvars.py
index e8634d0c..bfff5f74 100644
--- a/tests/async_fixtures/test_async_fixtures_contextvars.py
+++ b/tests/async_fixtures/test_async_fixtures_contextvars.py
@@ -101,7 +101,7 @@ def test(check_var_fixture):
         """
         )
     )
-    result = pytester.runpytest("--asyncio-mode=strict")
+    result = pytester.runpytest("--asyncio-mode=strict", "-s")
     result.assert_outcomes(passed=1)
 
 
diff --git a/tests/markers/test_invalid_arguments.py b/tests/markers/test_invalid_arguments.py
index 2d5c3552..fc2c88f1 100644
--- a/tests/markers/test_invalid_arguments.py
+++ b/tests/markers/test_invalid_arguments.py
@@ -40,9 +40,7 @@ async def test_anything():
     )
     result = pytester.runpytest_subprocess()
     result.assert_outcomes(errors=1)
-    result.stdout.fnmatch_lines(
-        ["*ValueError: mark.asyncio accepts only a keyword argument*"]
-    )
+    result.stdout.fnmatch_lines([""])
 
 
 def test_error_when_wrong_keyword_argument_is_passed(
@@ -62,7 +60,10 @@ async def test_anything():
     result = pytester.runpytest_subprocess()
     result.assert_outcomes(errors=1)
     result.stdout.fnmatch_lines(
-        ["*ValueError: mark.asyncio accepts only a keyword argument 'loop_scope'*"]
+        [
+            "*ValueError: mark.asyncio accepts only a keyword arguments "
+            "'loop_scope' or 'loop_factory'*"
+        ]
     )
 
 
@@ -83,5 +84,8 @@ async def test_anything():
     result = pytester.runpytest_subprocess()
     result.assert_outcomes(errors=1)
     result.stdout.fnmatch_lines(
-        ["*ValueError: mark.asyncio accepts only a keyword argument*"]
+        [
+            "*ValueError: mark.asyncio accepts only a keyword arguments "
+            "'loop_scope' or 'loop_factory'*"
+        ]
     )
diff --git a/tests/modes/test_strict_mode.py b/tests/modes/test_strict_mode.py
index d7dc4ac6..ecfd03da 100644
--- a/tests/modes/test_strict_mode.py
+++ b/tests/modes/test_strict_mode.py
@@ -163,10 +163,7 @@ async def test_anything(any_fixture):
     result.stdout.fnmatch_lines(
         [
             "*warnings summary*",
-            (
-                "test_strict_mode_marked_test_unmarked_fixture_warning.py::"
-                "test_anything"
-            ),
+            ("test_strict_mode_marked_test_unmarked_fixture_warning.py::test_anything"),
             (
                 "*/pytest_asyncio/plugin.py:*: PytestDeprecationWarning: "
                 "asyncio test 'test_anything' requested async "
diff --git a/tests/test_asyncio_mark.py b/tests/test_asyncio_mark.py
index 81731adb..26a12319 100644
--- a/tests/test_asyncio_mark.py
+++ b/tests/test_asyncio_mark.py
@@ -148,7 +148,7 @@ async def test_a():
     )
 
 
-def test_asyncio_marker_fallbacks_to_configured_default_loop_scope_if_not_set(
+def test_asyncio_marker_uses_marker_loop_scope_even_if_config_is_set(
     pytester: Pytester,
 ):
     pytester.makeini(
@@ -156,7 +156,7 @@ def test_asyncio_marker_fallbacks_to_configured_default_loop_scope_if_not_set(
             """\
             [pytest]
             asyncio_default_fixture_loop_scope = function
-            asyncio_default_test_loop_scope = session
+            asyncio_default_test_loop_scope = module
             """
         )
     )
@@ -175,6 +175,7 @@ async def session_loop_fixture():
                 global loop
                 loop = asyncio.get_running_loop()
 
+            @pytest.mark.asyncio(loop_scope="session")
             async def test_a(session_loop_fixture):
                 global loop
                 assert asyncio.get_running_loop() is loop
@@ -186,19 +187,162 @@ async def test_a(session_loop_fixture):
     result.assert_outcomes(passed=1)
 
 
-def test_asyncio_marker_uses_marker_loop_scope_even_if_config_is_set(
+def test_uses_loop_factory_from_test(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module")
+            async def any_fixture():
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+
+            @pytest.mark.asyncio(loop_scope="module", loop_factory=CustomEventLoop)
+            async def test_set_loop_factory(any_fixture):
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_uses_loop_factory_from_fixture(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def any_fixture():
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_set_loop_factory(any_fixture):
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_uses_loop_factory_from_transitive_fixture(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def transitive_fixture():
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+
+            @pytest_asyncio.fixture(loop_scope="module")
+            async def any_fixture(transitive_fixture):
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_set_loop_factory(any_fixture):
+                assert type(asyncio.get_running_loop()) == CustomEventLoop
+            """
+        )
+    )
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1)
+
+
+def test_conflicting_loop_factories_in_tests_raise_error(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            class AnotherCustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest.mark.asyncio(loop_scope="module", loop_factory=CustomEventLoop)
+            async def test_with_custom_loop_factory():
+                ...
+
+            @pytest.mark.asyncio(
+                loop_scope="module",
+                loop_factory=AnotherCustomEventLoop
+            )
+            async def test_with_a_different_custom_loop_factory():
+                ...
+            """
+        )
+    )
+
+    result = pytester.runpytest("--asyncio-mode=strict", "-s", "--setup-show")
+    result.assert_outcomes(errors=2)
+
+
+def test_conflicting_loop_factories_in_tests_and_fixtures_raise_error(
     pytester: Pytester,
 ):
-    pytester.makeini(
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
         dedent(
             """\
-            [pytest]
-            asyncio_default_fixture_loop_scope = function
-            asyncio_default_test_loop_scope = module
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            class AnotherCustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def fixture_with_custom_loop_factory():
+                ...
+
+            @pytest.mark.asyncio(
+                loop_scope="module",
+                loop_factory=AnotherCustomEventLoop
+            )
+            async def test_trying_to_override_fixtures_loop_factory(
+                fixture_with_custom_loop_factory
+            ):
+                # Fails, because it tries to use a different loop factory on the
+                # same runner as the first test
+                ...
             """
         )
     )
 
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(passed=1, errors=1)
+
+
+def test_conflicting_loop_factories_in_fixtures_raise_error(pytester: Pytester):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
     pytester.makepyfile(
         dedent(
             """\
@@ -206,20 +350,75 @@ def test_asyncio_marker_uses_marker_loop_scope_even_if_config_is_set(
             import pytest_asyncio
             import pytest
 
-            loop: asyncio.AbstractEventLoop
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
 
-            @pytest_asyncio.fixture(loop_scope="session", scope="session")
-            async def session_loop_fixture():
-                global loop
-                loop = asyncio.get_running_loop()
+            class AnotherCustomEventLoop(asyncio.SelectorEventLoop):
+                pass
 
-            @pytest.mark.asyncio(loop_scope="session")
-            async def test_a(session_loop_fixture):
-                global loop
-                assert asyncio.get_running_loop() is loop
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def fixture_with_custom_loop_factory():
+                ...
+
+            @pytest_asyncio.fixture(
+                loop_scope="module",
+                loop_factory=AnotherCustomEventLoop
+            )
+            async def another_fixture_with_custom_loop_factory():
+                ...
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_requesting_two_fixtures_with_different_loop_facoties(
+                fixture_with_custom_loop_factory,
+                another_fixture_with_custom_loop_factory,
+            ):
+                ...
             """
         )
     )
 
-    result = pytester.runpytest("--asyncio-mode=auto")
-    result.assert_outcomes(passed=1)
+    result = pytester.runpytest("--asyncio-mode=strict", "-s", "--setup-show")
+    result.assert_outcomes(errors=1)
+
+
+def test_conflicting_loop_factories_in_transitive_fixtures_raise_error(
+    pytester: Pytester,
+):
+    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function")
+    pytester.makepyfile(
+        dedent(
+            """\
+            import asyncio
+            import pytest_asyncio
+            import pytest
+
+            class CustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            class AnotherCustomEventLoop(asyncio.SelectorEventLoop):
+                pass
+
+            @pytest_asyncio.fixture(loop_scope="module", loop_factory=CustomEventLoop)
+            async def fixture_with_custom_loop_factory():
+                ...
+
+            @pytest_asyncio.fixture(
+                loop_scope="module",
+                loop_factory=AnotherCustomEventLoop
+            )
+            async def another_fixture_with_custom_loop_factory(
+                fixture_with_custom_loop_factory
+            ):
+                ...
+
+            @pytest.mark.asyncio(loop_scope="module")
+            async def test_requesting_two_fixtures_with_different_loop_facories(
+                another_fixture_with_custom_loop_factory,
+            ):
+                ...
+            """
+        )
+    )
+
+    result = pytester.runpytest("--asyncio-mode=strict")
+    result.assert_outcomes(errors=1)