diff --git a/homeassistant/components/mqtt/abbreviations.py b/homeassistant/components/mqtt/abbreviations.py
index c3f6b55e0fe..9cfa84c6c1b 100644
--- a/homeassistant/components/mqtt/abbreviations.py
+++ b/homeassistant/components/mqtt/abbreviations.py
@@ -8,6 +8,7 @@ ABBREVIATIONS = {
     "aux_stat_tpl": "aux_state_template",
     "aux_stat_t": "aux_state_topic",
     "avty": "availability",
+    "avty_mode": "availability_mode",
     "avty_t": "availability_topic",
     "away_mode_cmd_t": "away_mode_command_topic",
     "away_mode_stat_tpl": "away_mode_state_template",
diff --git a/homeassistant/components/mqtt/mixins.py b/homeassistant/components/mqtt/mixins.py
index e4fa7e15526..1ab2054b355 100644
--- a/homeassistant/components/mqtt/mixins.py
+++ b/homeassistant/components/mqtt/mixins.py
@@ -42,7 +42,14 @@ from .util import valid_subscribe_topic
 
 _LOGGER = logging.getLogger(__name__)
 
+AVAILABILITY_ALL = "all"
+AVAILABILITY_ANY = "any"
+AVAILABILITY_LATEST = "latest"
+
+AVAILABILITY_MODES = [AVAILABILITY_ALL, AVAILABILITY_ANY, AVAILABILITY_LATEST]
+
 CONF_AVAILABILITY = "availability"
+CONF_AVAILABILITY_MODE = "availability_mode"
 CONF_AVAILABILITY_TOPIC = "availability_topic"
 CONF_PAYLOAD_AVAILABLE = "payload_available"
 CONF_PAYLOAD_NOT_AVAILABLE = "payload_not_available"
@@ -71,6 +78,9 @@ MQTT_AVAILABILITY_SINGLE_SCHEMA = vol.Schema(
 
 MQTT_AVAILABILITY_LIST_SCHEMA = vol.Schema(
     {
+        vol.Optional(CONF_AVAILABILITY_MODE, default=AVAILABILITY_LATEST): vol.All(
+            cv.string, vol.In(AVAILABILITY_MODES)
+        ),
         vol.Exclusive(CONF_AVAILABILITY, "availability"): vol.All(
             cv.ensure_list,
             [
@@ -227,7 +237,8 @@ class MqttAvailability(Entity):
     def __init__(self, config: dict) -> None:
         """Initialize the availability mixin."""
         self._availability_sub_state = None
-        self._available = False
+        self._available = {}
+        self._available_latest = False
         self._availability_setup_from_config(config)
 
     async def async_added_to_hass(self) -> None:
@@ -275,12 +286,15 @@ class MqttAvailability(Entity):
             """Handle a new received MQTT availability message."""
             topic = msg.topic
             if msg.payload == self._avail_topics[topic][CONF_PAYLOAD_AVAILABLE]:
-                self._available = True
+                self._available[topic] = True
+                self._available_latest = True
             elif msg.payload == self._avail_topics[topic][CONF_PAYLOAD_NOT_AVAILABLE]:
-                self._available = False
+                self._available[topic] = False
+                self._available_latest = False
 
             self.async_write_ha_state()
 
+        self._available = {topic: False for topic in self._avail_topics}
         topics = {
             f"availability_{topic}": {
                 "topic": topic,
@@ -313,7 +327,13 @@ class MqttAvailability(Entity):
         """Return if the device is available."""
         if not self.hass.data[DATA_MQTT].connected and not self.hass.is_stopping:
             return False
-        return not self._avail_topics or self._available
+        if not self._avail_topics:
+            return True
+        if self._avail_config[CONF_AVAILABILITY_MODE] == AVAILABILITY_ALL:
+            return all(self._available.values())
+        if self._avail_config[CONF_AVAILABILITY_MODE] == AVAILABILITY_ANY:
+            return any(self._available.values())
+        return self._available_latest
 
 
 async def cleanup_device_registry(hass, device_id):
diff --git a/tests/components/mqtt/test_common.py b/tests/components/mqtt/test_common.py
index c5e271f1625..c8cd80372c6 100644
--- a/tests/components/mqtt/test_common.py
+++ b/tests/components/mqtt/test_common.py
@@ -171,6 +171,135 @@ async def help_test_default_availability_list_payload(
         assert state.state != STATE_UNAVAILABLE
 
 
+async def help_test_default_availability_list_payload_all(
+    hass,
+    mqtt_mock,
+    domain,
+    config,
+    no_assumed_state=False,
+    state_topic=None,
+    state_message=None,
+):
+    """Test availability by default payload with defined topic.
+
+    This is a test helper for the MqttAvailability mixin.
+    """
+    # Add availability settings to config
+    config = copy.deepcopy(config)
+    config[domain]["availability_mode"] = "all"
+    config[domain]["availability"] = [
+        {"topic": "availability-topic1"},
+        {"topic": "availability-topic2"},
+    ]
+    assert await async_setup_component(
+        hass,
+        domain,
+        config,
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state == STATE_UNAVAILABLE
+
+    async_fire_mqtt_message(hass, "availability-topic1", "online")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state == STATE_UNAVAILABLE
+    if no_assumed_state:
+        assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    async_fire_mqtt_message(hass, "availability-topic2", "online")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state != STATE_UNAVAILABLE
+
+    async_fire_mqtt_message(hass, "availability-topic2", "offline")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state == STATE_UNAVAILABLE
+    if no_assumed_state:
+        assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    async_fire_mqtt_message(hass, "availability-topic2", "online")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state != STATE_UNAVAILABLE
+
+    async_fire_mqtt_message(hass, "availability-topic1", "offline")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state == STATE_UNAVAILABLE
+    if no_assumed_state:
+        assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    async_fire_mqtt_message(hass, "availability-topic1", "online")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state != STATE_UNAVAILABLE
+
+
+async def help_test_default_availability_list_payload_any(
+    hass,
+    mqtt_mock,
+    domain,
+    config,
+    no_assumed_state=False,
+    state_topic=None,
+    state_message=None,
+):
+    """Test availability by default payload with defined topic.
+
+    This is a test helper for the MqttAvailability mixin.
+    """
+    # Add availability settings to config
+    config = copy.deepcopy(config)
+    config[domain]["availability_mode"] = "any"
+    config[domain]["availability"] = [
+        {"topic": "availability-topic1"},
+        {"topic": "availability-topic2"},
+    ]
+    assert await async_setup_component(
+        hass,
+        domain,
+        config,
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state == STATE_UNAVAILABLE
+
+    async_fire_mqtt_message(hass, "availability-topic1", "online")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state != STATE_UNAVAILABLE
+    if no_assumed_state:
+        assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    async_fire_mqtt_message(hass, "availability-topic2", "online")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state != STATE_UNAVAILABLE
+
+    async_fire_mqtt_message(hass, "availability-topic2", "offline")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state != STATE_UNAVAILABLE
+    if no_assumed_state:
+        assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    async_fire_mqtt_message(hass, "availability-topic1", "offline")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state == STATE_UNAVAILABLE
+
+    async_fire_mqtt_message(hass, "availability-topic1", "online")
+
+    state = hass.states.get(f"{domain}.test")
+    assert state.state != STATE_UNAVAILABLE
+    if no_assumed_state:
+        assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+
 async def help_test_default_availability_list_single(
     hass,
     mqtt_mock,
diff --git a/tests/components/mqtt/test_sensor.py b/tests/components/mqtt/test_sensor.py
index d818338a0ca..a2c2605d6ab 100644
--- a/tests/components/mqtt/test_sensor.py
+++ b/tests/components/mqtt/test_sensor.py
@@ -17,6 +17,8 @@ from .test_common import (
     help_test_availability_without_topic,
     help_test_custom_availability_payload,
     help_test_default_availability_list_payload,
+    help_test_default_availability_list_payload_all,
+    help_test_default_availability_list_payload_any,
     help_test_default_availability_list_single,
     help_test_default_availability_payload,
     help_test_discovery_broken,
@@ -297,6 +299,20 @@ async def test_default_availability_list_payload(hass, mqtt_mock):
     )
 
 
+async def test_default_availability_list_payload_all(hass, mqtt_mock):
+    """Test availability by default payload with defined topic."""
+    await help_test_default_availability_list_payload_all(
+        hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
+    )
+
+
+async def test_default_availability_list_payload_any(hass, mqtt_mock):
+    """Test availability by default payload with defined topic."""
+    await help_test_default_availability_list_payload_any(
+        hass, mqtt_mock, sensor.DOMAIN, DEFAULT_CONFIG
+    )
+
+
 async def test_default_availability_list_single(hass, mqtt_mock, caplog):
     """Test availability list and availability_topic are mutually exclusive."""
     await help_test_default_availability_list_single(