diff --git a/homeassistant/components/hikvision/__init__.py b/homeassistant/components/hikvision/__init__.py
index 2725c0dfba10..cc45552d5dc6 100644
--- a/homeassistant/components/hikvision/__init__.py
+++ b/homeassistant/components/hikvision/__init__.py
@@ -15,9 +15,10 @@ from homeassistant.const import (
     CONF_PORT,
     CONF_SSL,
     CONF_USERNAME,
+    EVENT_HOMEASSISTANT_STOP,
     Platform,
 )
-from homeassistant.core import HomeAssistant
+from homeassistant.core import Event, HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr
 
@@ -142,6 +143,21 @@ async def async_setup_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) ->
     # Start the event stream
     await hass.async_add_executor_job(camera.start_stream)
 
+    async def _async_stop_stream(event: Event) -> None:
+        await hass.async_add_executor_job(camera.disconnect)
+
+    # pyHik's stream thread is non-daemonic and publishes straight into hass, so
+    # it has to be joined before the event loop closes. Starting it yields, so
+    # the stop event may already have been fired by the time we get here, and
+    # listening for it now would never hear it.
+    if hass.is_stopping:
+        await hass.async_add_executor_job(camera.disconnect)
+        raise ConfigEntryNotReady("Home Assistant is stopping")
+
+    entry.async_on_unload(
+        hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_stop_stream)
+    )
+
     # Register the main device before platforms that use via_device
     device_registry = dr.async_get(hass)
     device_registry.async_get_or_create(
diff --git a/tests/components/hikvision/test_init.py b/tests/components/hikvision/test_init.py
index 7508dac4923e..05f30c11e7a2 100644
--- a/tests/components/hikvision/test_init.py
+++ b/tests/components/hikvision/test_init.py
@@ -8,8 +8,8 @@ import pytest
 import requests
 
 from homeassistant.config_entries import ConfigEntryState
-from homeassistant.const import CONF_SSL
-from homeassistant.core import HomeAssistant
+from homeassistant.const import CONF_SSL, EVENT_HOMEASSISTANT_STOP
+from homeassistant.core import CoreState, HomeAssistant
 
 from . import setup_integration
 from .conftest import TEST_HOST, TEST_PASSWORD, TEST_PORT, TEST_USERNAME
@@ -40,6 +40,36 @@ async def test_setup_and_unload_entry(
     mock_hikcamera.return_value.disconnect.assert_called_once()
 
 
+async def test_stream_stopped_on_shutdown(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_hikcamera: MagicMock,
+) -> None:
+    """Test the event stream thread is joined before the event loop closes."""
+    await setup_integration(hass, mock_config_entry)
+
+    hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
+    await hass.async_block_till_done()
+
+    mock_hikcamera.return_value.disconnect.assert_called_once()
+
+
+async def test_stream_stopped_when_shutdown_starts_during_setup(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_hikcamera: MagicMock,
+) -> None:
+    """Test the stream is stopped when shutdown begins while setting up."""
+    hass.set_state(CoreState.stopping)
+
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
+    mock_hikcamera.return_value.disconnect.assert_called_once()
+
+
 async def test_setup_entry_with_ssl(
     hass: HomeAssistant,
     mock_config_entry: MockConfigEntry,