diff --git a/homeassistant/components/mqtt/abbreviations.py b/homeassistant/components/mqtt/abbreviations.py
index 5e995494a64..6f9b1720102 100644
--- a/homeassistant/components/mqtt/abbreviations.py
+++ b/homeassistant/components/mqtt/abbreviations.py
@@ -135,6 +135,8 @@ ABBREVIATIONS = {
     "stat_off": "state_off",
     "stat_on": "state_on",
     "stat_open": "state_open",
+    "stat_locked": "state_locked",
+    "stat_unlocked": "state_unlocked",
     "stat_t": "state_topic",
     "stat_tpl": "state_template",
     "stat_val_tpl": "state_value_template",
diff --git a/homeassistant/components/mqtt/lock.py b/homeassistant/components/mqtt/lock.py
index ccf8f2569fa..6910e955288 100644
--- a/homeassistant/components/mqtt/lock.py
+++ b/homeassistant/components/mqtt/lock.py
@@ -36,10 +36,16 @@ _LOGGER = logging.getLogger(__name__)
 CONF_PAYLOAD_LOCK = "payload_lock"
 CONF_PAYLOAD_UNLOCK = "payload_unlock"
 
+CONF_STATE_LOCKED = "state_locked"
+CONF_STATE_UNLOCKED = "state_unlocked"
+
 DEFAULT_NAME = "MQTT Lock"
 DEFAULT_OPTIMISTIC = False
 DEFAULT_PAYLOAD_LOCK = "LOCK"
 DEFAULT_PAYLOAD_UNLOCK = "UNLOCK"
+DEFAULT_STATE_LOCKED = "LOCKED"
+DEFAULT_STATE_UNLOCKED = "UNLOCKED"
+
 PLATFORM_SCHEMA = (
     mqtt.MQTT_RW_PLATFORM_SCHEMA.extend(
         {
@@ -50,6 +56,10 @@ PLATFORM_SCHEMA = (
             vol.Optional(
                 CONF_PAYLOAD_UNLOCK, default=DEFAULT_PAYLOAD_UNLOCK
             ): cv.string,
+            vol.Optional(CONF_STATE_LOCKED, default=DEFAULT_STATE_LOCKED): cv.string,
+            vol.Optional(
+                CONF_STATE_UNLOCKED, default=DEFAULT_STATE_UNLOCKED
+            ): cv.string,
             vol.Optional(CONF_UNIQUE_ID): cv.string,
         }
     )
@@ -152,9 +162,9 @@ class MqttLock(
             payload = msg.payload
             if value_template is not None:
                 payload = value_template.async_render_with_possible_json_value(payload)
-            if payload == self._config[CONF_PAYLOAD_LOCK]:
+            if payload == self._config[CONF_STATE_LOCKED]:
                 self._state = True
-            elif payload == self._config[CONF_PAYLOAD_UNLOCK]:
+            elif payload == self._config[CONF_STATE_UNLOCKED]:
                 self._state = False
 
             self.async_write_ha_state()
diff --git a/tests/components/mqtt/test_lock.py b/tests/components/mqtt/test_lock.py
index fbaedd3f945..9b89fa7159d 100644
--- a/tests/components/mqtt/test_lock.py
+++ b/tests/components/mqtt/test_lock.py
@@ -34,6 +34,8 @@ async def test_controlling_state_via_topic(hass, mqtt_mock):
                 "command_topic": "command-topic",
                 "payload_lock": "LOCK",
                 "payload_unlock": "UNLOCK",
+                "state_locked": "LOCKED",
+                "state_unlocked": "UNLOCKED",
             }
         },
     )
@@ -42,12 +44,46 @@ async def test_controlling_state_via_topic(hass, mqtt_mock):
     assert state.state is STATE_UNLOCKED
     assert not state.attributes.get(ATTR_ASSUMED_STATE)
 
-    async_fire_mqtt_message(hass, "state-topic", "LOCK")
+    async_fire_mqtt_message(hass, "state-topic", "LOCKED")
 
     state = hass.states.get("lock.test")
     assert state.state is STATE_LOCKED
 
-    async_fire_mqtt_message(hass, "state-topic", "UNLOCK")
+    async_fire_mqtt_message(hass, "state-topic", "UNLOCKED")
+
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_UNLOCKED
+
+
+async def test_controlling_non_default_state_via_topic(hass, mqtt_mock):
+    """Test the controlling state via topic."""
+    assert await async_setup_component(
+        hass,
+        lock.DOMAIN,
+        {
+            lock.DOMAIN: {
+                "platform": "mqtt",
+                "name": "test",
+                "state_topic": "state-topic",
+                "command_topic": "command-topic",
+                "payload_lock": "LOCK",
+                "payload_unlock": "UNLOCK",
+                "state_locked": "closed",
+                "state_unlocked": "open",
+            }
+        },
+    )
+
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_UNLOCKED
+    assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    async_fire_mqtt_message(hass, "state-topic", "closed")
+
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_LOCKED
+
+    async_fire_mqtt_message(hass, "state-topic", "open")
 
     state = hass.states.get("lock.test")
     assert state.state is STATE_UNLOCKED
@@ -66,6 +102,8 @@ async def test_controlling_state_via_topic_and_json_message(hass, mqtt_mock):
                 "command_topic": "command-topic",
                 "payload_lock": "LOCK",
                 "payload_unlock": "UNLOCK",
+                "state_locked": "LOCKED",
+                "state_unlocked": "UNLOCKED",
                 "value_template": "{{ value_json.val }}",
             }
         },
@@ -74,12 +112,48 @@ async def test_controlling_state_via_topic_and_json_message(hass, mqtt_mock):
     state = hass.states.get("lock.test")
     assert state.state is STATE_UNLOCKED
 
-    async_fire_mqtt_message(hass, "state-topic", '{"val":"LOCK"}')
+    async_fire_mqtt_message(hass, "state-topic", '{"val":"LOCKED"}')
 
     state = hass.states.get("lock.test")
     assert state.state is STATE_LOCKED
 
-    async_fire_mqtt_message(hass, "state-topic", '{"val":"UNLOCK"}')
+    async_fire_mqtt_message(hass, "state-topic", '{"val":"UNLOCKED"}')
+
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_UNLOCKED
+
+
+async def test_controlling_non_default_state_via_topic_and_json_message(
+    hass, mqtt_mock
+):
+    """Test the controlling state via topic and JSON message."""
+    assert await async_setup_component(
+        hass,
+        lock.DOMAIN,
+        {
+            lock.DOMAIN: {
+                "platform": "mqtt",
+                "name": "test",
+                "state_topic": "state-topic",
+                "command_topic": "command-topic",
+                "payload_lock": "LOCK",
+                "payload_unlock": "UNLOCK",
+                "state_locked": "closed",
+                "state_unlocked": "open",
+                "value_template": "{{ value_json.val }}",
+            }
+        },
+    )
+
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_UNLOCKED
+
+    async_fire_mqtt_message(hass, "state-topic", '{"val":"closed"}')
+
+    state = hass.states.get("lock.test")
+    assert state.state is STATE_LOCKED
+
+    async_fire_mqtt_message(hass, "state-topic", '{"val":"open"}')
 
     state = hass.states.get("lock.test")
     assert state.state is STATE_UNLOCKED
@@ -97,6 +171,8 @@ async def test_sending_mqtt_commands_and_optimistic(hass, mqtt_mock):
                 "command_topic": "command-topic",
                 "payload_lock": "LOCK",
                 "payload_unlock": "UNLOCK",
+                "state_locked": "LOCKED",
+                "state_unlocked": "UNLOCKED",
             }
         },
     )
@@ -135,6 +211,8 @@ async def test_sending_mqtt_commands_and_explicit_optimistic(hass, mqtt_mock):
                 "command_topic": "command-topic",
                 "payload_lock": "LOCK",
                 "payload_unlock": "UNLOCK",
+                "state_locked": "LOCKED",
+                "state_unlocked": "UNLOCKED",
                 "optimistic": True,
             }
         },
@@ -206,6 +284,8 @@ async def test_custom_availability_payload(hass, mqtt_mock):
                 "command_topic": "command-topic",
                 "payload_lock": "LOCK",
                 "payload_unlock": "UNLOCK",
+                "state_locked": "LOCKED",
+                "state_unlocked": "UNLOCKED",
                 "availability_topic": "availability-topic",
                 "payload_available": "good",
                 "payload_not_available": "nogood",