AWS Lambda解码MQTT消息触发UTF-8无效字节解码错误求助

你触发UTF-8解码报错的本质是：经过Base64解码得到的字节序列本身不符合UTF-8编码规则。
UTF-8编码有严格的格式规范：单字节字符最高位为0（对应0x00-0x7F的ASCII范围），多字节字符的首字节会标记字节长度，后续所有字节必须落在0x80-0xBF区间。你给出的复现案例中，解码得到的字节串b'E\xa9\xdd\xa2d\xde\xc6'里，首字节0x45是ASCII字符'E'，下一字节0xa9属于多字节后续字节的取值范围，前面没有匹配的多字节首字节，属于非法UTF-8序列，自然解码失败。
你用的测试入参本身就不符合逻辑："Random Text=="本身是可读明文，根本不是UTF-8文本对应的合法Base64编码结果。如果要把明文Random Text转成Base64，正确编码结果是UmFuZG9tIFRleHQ=，解码后才能得到合法的UTF-8字节。

• 先对齐上游MQTT消息的生产逻辑
  90%以上的这类问题都是上下游编码逻辑不匹配导致的：
  • 要么上游根本没有对消息做Base64编码，直接把明文塞到data字段，你多余做了Base64解码
  • 要么上游传的不是UTF-8编码的文本，比如是GBK/Latin1编码的内容、传感器原始二进制字节流、压缩/加密后的二进制数据，本身就不能直接按UTF-8转字符串
• 对应调整Lambda处理逻辑
  • 如果上游直接传明文JSON，没有做Base64编码：直接删除Base64解码、UTF-8解码步骤，直接读取event['data']字段使用即可
  • 如果上游传的是其他编码的文本：将UTF-8替换为实际编码做解码，比如GBK编码用eventText.decode("GBK")，Latin1编码用eventText.decode("ISO-8859-1")
  • 如果上游传的是自定义二进制业务数据：不要强行转字符串，直接按二进制格式解析对应业务字段即可
• 增加异常兜底避免函数报错
  MQTT消息可能存在异常上报的情况，建议增加异常捕获和兼容逻辑，参考代码如下：

import base64

def lambda_handler(event, context):
    try:
        raw_data = event.get('data', '')
        # 可根据自身业务规则判断是否需要做Base64解码，避免对明文误解码
        if isinstance(raw_data, str) and len(raw_data) % 4 == 0 and all(c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=' for c in raw_data):
            content_bytes = base64.b64decode(raw_data)
        else:
            content_bytes = raw_data.encode('utf-8') if isinstance(raw_data, str) else raw_data
        
        # 优先按UTF-8解码，失败时尝试常见编码，异常字符用替换符兜底
        content = None
        for codec in ['utf-8', 'gbk', 'iso-8859-1']:
            try:
                content = content_bytes.decode(codec)
                break
            except UnicodeDecodeError:
                continue
        if not content:
            content = content_bytes.decode('utf-8', errors='replace')
        
        # 此处编写后续业务处理逻辑
        return {"code": 0, "data": content}
    except Exception as e:
        print(f"消息处理异常，入参：{event}，错误信息：{str(e)}")
        return {"code": -1, "error": str(e)}

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