在Alexa AWS Lambda函数中集成MQTT时消息发布不稳定求助

我尝试让Alexa技能通过Lambda代码向MQTT Broker发布消息，但消息仅偶尔能发布成功。调用Alexa技能能收到正常响应，但发布行为不稳定。代码如下：

const Alexa = require('ask-sdk-core');
const mqtt = require("mqtt");


const protocol = 'mqtt'
const host = 'm21.cloudmqtt.com'
const port = '14297'
const clientId = `mqtt_${Math.random().toString(16).slice(3)}`

const connectUrl = `${protocol}://${host}:${port}`
console.log(connectUrl);
const client = mqtt.connect(connectUrl, {
  clientId,
  clean: false,
  connectTimeout: 4000,
  username: '****',
  password: '***',
  reconnectPeriod: 1000,
})
client.on('connect',  function(connack) { 
        console.log("MQTT Client connected " + JSON.stringify(connack));
        // publish a message to a topic
        client.publish('/IoTmanager/YardESP/Garage_close3/control', '{"status":0}', function() {
            console.log("Message is published");
            client.end(); // Close the connection when published
        });

        });
client.on('end', () => { 
  console.log('Connection to MQTT broker ended');
});
//
const LaunchRequestHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'LaunchRequest';
  },
  handle(handlerInput) {
    const speechText = 'Would you like to open ur shut the Garage Door?';
    console.log("MQTT Connecting");
    /*client.on("connect", () => {
        console.log("MQTT Connected");
        client.publish("presence", "Hello mqtt");
    });*/

    //
    return handlerInput.responseBuilder
      .speak(speechText)
      .reprompt(speechText)
      .withSimpleCard('Would you like to open ur shut the Garage Door?', speechText)
      .getResponse();
  }
};
const GarageDoorCloseIntentHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
      && Alexa.getIntentName(handlerInput.requestEnvelope) === 'GarageDoorCloseIntent';
  },
  handle(handlerInput) {
    const speechText = 'Closing The garage door!';
    console.log("MQTT Options: ", speechText);
    return handlerInput.responseBuilder
      .speak(speechText)
      .withSimpleCard('Closing The garage door!', speechText)
      .getResponse();
  }
};
const GarageDoorOpenIntentHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
      && Alexa.getIntentName(handlerInput.requestEnvelope) === 'GarageDoorOpenIntent';
  },
  handle(handlerInput) {
    const speechText = 'Opening The garage door!';

    return handlerInput.responseBuilder
      .speak(speechText)
      .withSimpleCard('Opening The garage door!', speechText)
      .getResponse();
  }
};
const HelpIntentHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
      && Alexa.getIntentName(handlerInput.requestEnvelope) === 'AMAZON.HelpIntent';
  },
  handle(handlerInput) {
    const speechText = 'You can ask me the weather!';

    return handlerInput.responseBuilder
      .speak(speechText)
      .reprompt(speechText)
      .withSimpleCard('You can ask me the weather!', speechText)
      .getResponse();
  }
};
const CancelAndStopIntentHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
      && (Alexa.getIntentName(handlerInput.requestEnvelope) === 'AMAZON.CancelIntent'
        || Alexa.getIntentName(handlerInput.requestEnvelope) === 'AMAZON.StopIntent');
  },
  handle(handlerInput) {
    const speechText = 'Goodbye!';

    return handlerInput.responseBuilder
      .speak(speechText)
      .withSimpleCard('Goodbye!', speechText)
      .withShouldEndSession(true)
      .getResponse();
  }
};
const SessionEndedRequestHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'SessionEndedRequest';
  },
  handle(handlerInput) {
    // Any clean-up logic goes here.
    return handlerInput.responseBuilder.getResponse();
  }
};
const ErrorHandler = {
  canHandle() {
    return true;
  },
  handle(handlerInput, error) {
    console.log(`Error handled: ${error.message}`);

    return handlerInput.responseBuilder
      .speak('Sorry, I don\'t understand your command. Please say it again.')
      .reprompt('Sorry, I don\'t understand your command. Please say it again.')
      .getResponse();
  }
};
let skill;

exports.handler = async function (event, context) {
  console.log(`REQUEST++++${JSON.stringify(event)}`);
  console.log(context);
  if (!skill) {
    skill = Alexa.SkillBuilders.custom()
      .addRequestHandlers(
        LaunchRequestHandler,
        GarageDoorCloseIntentHandler,
        GarageDoorOpenIntentHandler,
        HelpIntentHandler,
        CancelAndStopIntentHandler,
        SessionEndedRequestHandler,
      )
      .addErrorHandlers(ErrorHandler)
      .create();
  }


  const response = await skill.invoke(event, context);
  console.log(`RESPONSE++++${JSON.stringify(response)}`);

  return response;
};

问题分析 #

你的代码存在几个关键问题，导致MQTT发布行为不稳定：

1. MQTT逻辑位置错误：当前MQTT连接和发布代码写在Lambda的全局作用域里，只有Lambda容器冷启动时会执行一次。后续请求复用容器时，这段代码不会重复执行，而且你在connect回调里发布消息后直接调用client.end()关闭连接，后续请求根本没有可用的连接来发布消息。
2. Intent处理未触发MQTT发布：你的GarageDoorCloseIntentHandler和GarageDoorOpenIntentHandler里只返回了语音响应，完全没有调用MQTT发布消息的逻辑。之前偶尔成功的情况，只是刚好在Lambda冷启动时触发了全局作用域里的发布代码，和你调用技能意图的操作无关。
3. 缺少错误处理：没有监听MQTT客户端的error事件，无法排查连接或发布失败的原因。

修复方案 #

把MQTT连接和发布逻辑封装成异步函数，在对应的Intent处理函数里调用，确保每次触发开门/关门意图时都执行发布操作。同时添加错误处理，确保连接失败时能捕获异常。

修复后的代码示例 #

const Alexa = require('ask-sdk-core');
const mqtt = require("mqtt");

// 封装MQTT发布函数
async function publishMQTTMessage(topic, message) {
  return new Promise((resolve, reject) => {
    const protocol = 'mqtt';
    const host = 'm21.cloudmqtt.com';
    const port = '14297';
    const clientId = `mqtt_${Math.random().toString(16).slice(3)}`;
    const connectUrl = `${protocol}://${host}:${port}`;

    const client = mqtt.connect(connectUrl, {
      clientId,
      clean: true, // 改为clean: true，避免会话残留
      connectTimeout: 4000,
      username: '****',
      password: '***',
      reconnectPeriod: 1000,
    });

    // 监听连接成功事件
    client.on('connect', () => {
      console.log("MQTT Client connected");
      client.publish(topic, message, (err) => {
        if (err) {
          console.error("MQTT publish error:", err);
          reject(err);
        } else {
          console.log("Message is published");
          resolve();
        }
        client.end(); // 发布后关闭连接
      });
    });

    // 监听连接错误事件
    client.on('error', (err) => {
      console.error("MQTT connection error:", err);
      reject(err);
      client.end();
    });

    // 监听超时
    setTimeout(() => {
      reject(new Error("MQTT connection timeout"));
      client.end();
    }, 5000);
  });
}

const LaunchRequestHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'LaunchRequest';
  },
  handle(handlerInput) {
    const speechText = 'Would you like to open or shut the Garage Door?';
    return handlerInput.responseBuilder
      .speak(speechText)
      .reprompt(speechText)
      .withSimpleCard('Garage Door Control', speechText)
      .getResponse();
  }
};

const GarageDoorCloseIntentHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
      && Alexa.getIntentName(handlerInput.requestEnvelope) === 'GarageDoorCloseIntent';
  },
  async handle(handlerInput) {
    let speechText = 'Closing the garage door!';
    try {
      await publishMQTTMessage('/IoTmanager/YardESP/Garage_close3/control', '{"status":0}');
    } catch (err) {
      speechText = 'Sorry, failed to close the garage door.';
      console.error("Close door MQTT error:", err);
    }
    return handlerInput.responseBuilder
      .speak(speechText)
      .withSimpleCard('Garage Door Control', speechText)
      .getResponse();
  }
};

const GarageDoorOpenIntentHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
      && Alexa.getIntentName(handlerInput.requestEnvelope) === 'GarageDoorOpenIntent';
  },
  async handle(handlerInput) {
    let speechText = 'Opening the garage door!';
    try {
      // 替换为开门对应的topic和消息
      await publishMQTTMessage('/IoTmanager/YardESP/Garage_open3/control', '{"status":1}');
    } catch (err) {
      speechText = 'Sorry, failed to open the garage door.';
      console.error("Open door MQTT error:", err);
    }
    return handlerInput.responseBuilder
      .speak(speechText)
      .withSimpleCard('Garage Door Control', speechText)
      .getResponse();
  }
};

// 其他Handler保持不变
const HelpIntentHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
      && Alexa.getIntentName(handlerInput.requestEnvelope) === 'AMAZON.HelpIntent';
  },
  handle(handlerInput) {
    const speechText = 'You can ask me to open or close the garage door!';
    return handlerInput.responseBuilder
      .speak(speechText)
      .reprompt(speechText)
      .withSimpleCard('Garage Door Control Help', speechText)
      .getResponse();
  }
};

const CancelAndStopIntentHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'IntentRequest'
      && (Alexa.getIntentName(handlerInput.requestEnvelope) === 'AMAZON.CancelIntent'
        || Alexa.getIntentName(handlerInput.requestEnvelope) === 'AMAZON.StopIntent');
  },
  handle(handlerInput) {
    const speechText = 'Goodbye!';
    return handlerInput.responseBuilder
      .speak(speechText)
      .withSimpleCard('Garage Door Control', speechText)
      .withShouldEndSession(true)
      .getResponse();
  }
};

const SessionEndedRequestHandler = {
  canHandle(handlerInput) {
    return Alexa.getRequestType(handlerInput.requestEnvelope) === 'SessionEndedRequest';
  },
  handle(handlerInput) {
    return handlerInput.responseBuilder.getResponse();
  }
};

const ErrorHandler = {
  canHandle() {
    return true;
  },
  handle(handlerInput, error) {
    console.log(`Error handled: ${error.message}`);
    return handlerInput.responseBuilder
      .speak('Sorry, I don\'t understand your command. Please say it again.')
      .reprompt('Sorry, I don\'t understand your command. Please say it again.')
      .getResponse();
  }
};

let skill;

exports.handler = async function (event, context) {
  console.log(`REQUEST++++${JSON.stringify(event)}`);
  if (!skill) {
    skill = Alexa.SkillBuilders.custom()
      .addRequestHandlers(
        LaunchRequestHandler,
        GarageDoorCloseIntentHandler,
        GarageDoorOpenIntentHandler,
        HelpIntentHandler,
        CancelAndStopIntentHandler,
        SessionEndedRequestHandler,
      )
      .addErrorHandlers(ErrorHandler)
      .create();
  }
  const response = await skill.invoke(event, context);
  console.log(`RESPONSE++++${JSON.stringify(response)}`);
  return response;
};

修复说明 #

• 把MQTT发布逻辑封装成异步函数publishMQTTMessage，每次调用意图时都会创建新的连接并发布消息，避免容器复用导致的问题。
• 在Intent处理函数中使用async/await等待MQTT操作完成，确保发布结果能反馈到语音响应中。
• 添加了错误处理，连接或发布失败时会捕获异常，并返回对应的错误语音提示。
• 将clean选项改为true，避免客户端会话残留导致的连接问题。
• 给MQTT操作添加了超时处理，防止Lambda因等待过久而超时。

内容的提问来源于stack exchange，提问作者Murray Clark