diff --git a/homeassistant/components/esphome/__init__.py b/homeassistant/components/esphome/__init__.py
index 59db885d450..192a19e480b 100644
--- a/homeassistant/components/esphome/__init__.py
+++ b/homeassistant/components/esphome/__init__.py
@@ -345,6 +345,10 @@ async def async_setup_entry(  # noqa: C901
             disconnect_cb()
         entry_data.disconnect_callbacks = []
         entry_data.available = False
+        # Clear out the states so that we will always dispatch
+        # the next state update of that type when the device reconnects
+        for state_keys in entry_data.state.values():
+            state_keys.clear()
         entry_data.async_update_device_state(hass)
 
     async def on_connect_error(err: Exception) -> None:
@@ -760,7 +764,7 @@ class EsphomeEntity(Entity, Generic[_InfoT, _StateT]):
         self.async_on_remove(
             async_dispatcher_connect(
                 self.hass,
-                f"esphome_{self._entry_id}_on_device_update",
+                self._entry_data.signal_device_updated,
                 self._on_device_update,
             )
         )
diff --git a/homeassistant/components/esphome/entry_data.py b/homeassistant/components/esphome/entry_data.py
index 0aed6ce43a7..4d035427085 100644
--- a/homeassistant/components/esphome/entry_data.py
+++ b/homeassistant/components/esphome/entry_data.py
@@ -39,6 +39,7 @@ from homeassistant.helpers.storage import Store
 
 from .dashboard import async_get_dashboard
 
+_SENTINEL = object()
 SAVE_DELAY = 120
 _LOGGER = logging.getLogger(__name__)
 
@@ -198,14 +199,26 @@ class RuntimeEntryData:
     @callback
     def async_update_state(self, state: EntityState) -> None:
         """Distribute an update of state information to the target."""
-        subscription_key = (type(state), state.key)
-        self.state[type(state)][state.key] = state
+        key = state.key
+        state_type = type(state)
+        current_state_by_type = self.state[state_type]
+        current_state = current_state_by_type.get(key, _SENTINEL)
+        if current_state == state:
+            _LOGGER.debug(
+                "%s: ignoring duplicate update with and key %s: %s",
+                self.name,
+                key,
+                state,
+            )
+            return
         _LOGGER.debug(
             "%s: dispatching update with key %s: %s",
             self.name,
-            subscription_key,
+            key,
             state,
         )
+        current_state_by_type[key] = state
+        subscription_key = (state_type, key)
         if subscription_key in self.state_subscriptions:
             self.state_subscriptions[subscription_key]()