Python循环仅识别首个JSON对象，所有数据发布到同一MQTT主题如何解决？

• 存在冗余嵌套循环：你已经在json_data中存储了每个设备的完整配置信息，不需要单独提取uuid_list再做两层嵌套遍历。当前外层每遍历1个uuid，内层都会完整遍历全部5条配置，MQTT_TOPIC会被反复覆盖，最终保留的是内层循环最后一次迭代的配置值，和当前处理的uuid完全不匹配
• 发布逻辑缩进错误：消息发布的for循环被放在了内层遍历json_data的循环之外，发布时调用的MQTT_TOPIC已经不是当前uuid对应的值
• 重复创建MQTT连接：每次内层循环都会新建一个MQTT连接，完全冗余，也会导致最后使用的连接和主题不匹配

直接遍历json_data的每条设备配置，同时通过索引匹配预生成的模拟数据即可：

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0
from awscrt import io, mqtt, auth, http
from awsiot import mqtt_connection_builder
import time as t
import json
import os, tempfile
import boto3
from datetime import datetime
import pandas as pd
import random
import time

s3 = boto3.resource('s3')
client = boto3.client('s3')
data_bucket = '3deo-sensor-data'
key = 'simulated/config/IoT-sim-config.json'
obj = s3.Object(data_bucket, key)
data = obj.get()['Body'].read().decode('utf-8')
json_data = json.loads(data)

pd.set_option('display.max_columns', None)
pd.set_option('display.max_rows', None)
time_format = '%m/%d/%Y %I:%M:%S %p'
RANGE = 2

# 预生成所有设备的模拟数据
uuid_list = [item["uuid"] for item in json_data]
def AQ_simulator():
    d = datetime.now().strftime(time_format)
    ts = time.strptime(d, '%m/%d/%Y  %I:%M:%S %p')
    ts1 = time.mktime(ts)
    timestamp = int(ts1)
    df = pd.DataFrame()
    for uuid in uuid_list: 
        for x in range(RANGE):
            temp = pd.DataFrame({
                    'uuid': uuid,
                    'timestamp': str(timestamp),
                    'NO2 concentration': round(random.uniform(0, 0.017), 3),
                    'SO2 concentration': round(random.uniform(0, 0.022), 3),
                    'temperature': str(random.randint(3, 18)),
                    'relative humidity': str(random.randint(75, 90)),
                    'air pressure': str(random.randint(1011, 1016)),
                    'PM1.0 concentration': str(random.randint(5, 50)),
                    'PM2.5 concentration': str(random.randint(11, 26)),
                    'PM4 concentration': str(random.randint(7, 22)),
                    'PM10 concentration': str(random.randint(1, 20))[:4]}, index=[0],
                    columns = ['uuid', 'timestamp', 'NO2 concentration', 'SO2 concentration', 'temperature', 'relative humidity', 'air pressure', 'PM1.0 concentration', 'PM2.5 concentration', 'PM4 concentration', 'PM10 concentration'])
            df = pd.concat([df, temp])
    result = df.to_json(orient="records")
    parsed = json.loads(result)
    return parsed

MESSAGE = AQ_simulator()

if 'ENDPOINT' in os.environ:
    ENDPOINT = os.environ['ENDPOINT']
else:
    ENDPOINT = 'a3dd26595zqo4r-ats.iot.eu-west-1.amazonaws.com'
if 'BUCKET ' in os.environ:
    BUCKET = os.environ['BUCKET']
else:
    BUCKET = '3deo-sensor-certificates'

def lambda_handler(event, context):
    print("\nvvvvvvvvvvvvvvv")
    print(event)
    print("^^^^^^^^^^^^^^\n")

cert_bucket = '3deo-sensor-certificates'

# 直接遍历每条设备配置，同时记录索引匹配模拟数据
for device_index, device_config in enumerate(json_data):
    # 直接从当前配置取所有参数，不需要嵌套循环
    CLIENT_ID = device_config["client_id"]
    MQTT_TOPIC = f"{device_config['organisation']}/{device_config['device_vendor']}/{device_config['device_type']}"
    CERTIFICATION_PATH = f"{device_config['organisation']}/{device_config['uuid']}"
    result = client.list_objects(Bucket=cert_bucket, Prefix=CERTIFICATION_PATH)
    print('-------------------')
    print(result)
    print('-------------------')
        
    list_out = []
    if 'Contents' in result:
        for key in result['Contents']:
            if not key['Key'].endswith('/'):
                list_out.append(key['Key'])
        print(list_out)
    
    with tempfile.TemporaryDirectory() as tmpdir:
        os.chdir(tmpdir)
        PATH_TO_CERT = None
        PATH_TO_KEY = None
        PATH_TO_ROOT = None
        for item in list_out:
            if 'certificate.pem.crt' in item:
                cert_file = item.rsplit('/')[-1]
                client.download_file(cert_bucket, item, cert_file)
                PATH_TO_CERT = os.path.join(os.getcwd(), cert_file)
            elif 'private.pem.key' in item:
                key_file = item.rsplit('/')[-1]
                client.download_file(cert_bucket, item, key_file)
                PATH_TO_KEY = os.path.join(os.getcwd(), key_file)
            elif 'AmazonRootCA1.pem' in item:
                root_file = item.rsplit('/')[-1]
                client.download_file(cert_bucket, item, root_file)
                PATH_TO_ROOT = os.path.join(os.getcwd(), root_file)
        print(PATH_TO_CERT)
        print(PATH_TO_KEY)
        print(PATH_TO_ROOT)
        
        # 建立MQTT连接
        event_loop_group = io.EventLoopGroup(1)
        host_resolver = io.DefaultHostResolver(event_loop_group)
        client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)
        mqtt_connection = mqtt_connection_builder.mtls_from_path(
            endpoint=ENDPOINT,
            cert_filepath=PATH_TO_CERT,
            pri_key_filepath=PATH_TO_KEY,
            client_bootstrap=client_bootstrap,
            ca_filepath=PATH_TO_ROOT,
            client_id=CLIENT_ID,
            clean_session=False,
            keep_alive_secs=6
        )
        print(f"Connecting to {ENDPOINT} with client ID '{CLIENT_ID}'...")
        connect_future = mqtt_connection.connect()
        connect_future.result()
        print("Connected!")
        
        # 发布当前设备对应的模拟数据
        print('Begin Publish')
        for data in MESSAGE[device_index * RANGE : (device_index+1) * RANGE]:
            mqtt_connection.publish(topic=MQTT_TOPIC, payload=json.dumps(data), qos=mqtt.QoS.AT_LEAST_ONCE)
            print(f"Published: '{json.dumps(data)}' to the topic: {MQTT_TOPIC}")
            t.sleep(3)
        
        # 处理完当前设备的消息后断开连接
        disconnect_future = mqtt_connection.disconnect()
        disconnect_future.result()

print('Publish End')

调整后每个设备的模拟数据都会使用自身配置对应的MQTT主题发布，不会出现所有消息都发到同一个主题的问题。

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