diff --git a/homeassistant/components/mqtt/__init__.py b/homeassistant/components/mqtt/__init__.py
index ff99cf59c0b90..a7ced0f80513f 100644
--- a/homeassistant/components/mqtt/__init__.py
+++ b/homeassistant/components/mqtt/__init__.py
@@ -872,7 +872,9 @@ async def async_subscribe(
         subscription = Subscription(topic, msg_callback, qos, encoding)
         self.subscriptions.append(subscription)
 
-        await self._async_perform_subscription(topic, qos)
+        # Only subscribe if currently connected.
+        if self.connected:
+            await self._async_perform_subscription(topic, qos)
 
         @callback
         def async_remove() -> None:
diff --git a/tests/components/mqtt/test_init.py b/tests/components/mqtt/test_init.py
index e70a53caf9d2f..997a134fc8e91 100644
--- a/tests/components/mqtt/test_init.py
+++ b/tests/components/mqtt/test_init.py
@@ -608,6 +608,9 @@ def side_effect(*args):
 
         self.hass.data["mqtt"]._mqttc.subscribe.side_effect = side_effect
 
+        # Fake that the client is connected
+        self.hass.data["mqtt"].connected = True
+
         calls_a = mock.MagicMock()
         mqtt.subscribe(self.hass, "test/state", calls_a)
         self.hass.block_till_done()
@@ -620,6 +623,9 @@ def side_effect(*args):
 
     def test_not_calling_unsubscribe_with_active_subscribers(self):
         """Test not calling unsubscribe() when other subscribers are active."""
+        # Fake that the client is connected
+        self.hass.data["mqtt"].connected = True
+
         unsub = mqtt.subscribe(self.hass, "test/state", None)
         mqtt.subscribe(self.hass, "test/state", None)
         self.hass.block_till_done()
@@ -631,6 +637,9 @@ def test_not_calling_unsubscribe_with_active_subscribers(self):
 
     def test_restore_subscriptions_on_reconnect(self):
         """Test subscriptions are restored on reconnect."""
+        # Fake that the client is connected
+        self.hass.data["mqtt"].connected = True
+
         mqtt.subscribe(self.hass, "test/state", None)
         self.hass.block_till_done()
         assert self.hass.data["mqtt"]._mqttc.subscribe.call_count == 1
@@ -642,6 +651,9 @@ def test_restore_subscriptions_on_reconnect(self):
 
     def test_restore_all_active_subscriptions_on_reconnect(self):
         """Test active subscriptions are restored correctly on reconnect."""
+        # Fake that the client is connected
+        self.hass.data["mqtt"].connected = True
+
         self.hass.data["mqtt"]._mqttc.subscribe.side_effect = (
             (0, 1),
             (0, 2),