流订阅处理程序单输入行转多行追加时数据丢失问题排查

1. 未初始化变量引发异常中断
   代码中tb_ctp_pre_price仅在SecurityID匹配四个特定值时才会被赋值，若输入的SecurityID不在这四个值中，该变量未定义，后续执行update tb_ctp_pre_price会直接抛出异常，导致当前处理逻辑中断，甚至阻塞后续所有流表消息的处理，这是大部分数据未被转换的核心原因。

2. 空表访问触发隐性异常
   当msg经过过滤条件where (not isNull(lastPrice)) and lastPrice != 0后得到的temp为空表时，执行temp['SecurityID'][0]会触发数组越界异常，直接终止处理流程，导致该行数据完全未被处理。

3. 流表处理的异常传播
   流表的订阅处理逻辑通常是单线程串行执行的，一旦某次处理抛出未捕获的异常，后续的消息处理会被中断或跳过，这就解释了为什么只有前几行或偶尔的行能成功处理。

1. 确保变量始终初始化 #

给tb_ctp_pre_price添加默认赋值，同时将多个独立if改为elif减少不必要的判断：

def handleCtp_ETF_Subs(mutable msg, hqTbName, reorderedColNames,csv_priceData) {
    temp = select securityCode as SecurityID,preClosePrice,lastPrice as Price,origTime as tradetime,receivedTime from msg where (not isNull(lastPrice)) and lastPrice != 0
    update temp set chg_pct= (double(Price)/double(preClosePrice))-1 

    // 初始化空表，避免变量未定义
    tb_ctp_pre_price = table(0:0, ['SecurityID', 'Price', 'tradetime', 'receivedTime'], [STRING, DOUBLE, DATETIME, DATETIME])
    if(temp['SecurityID'][0] == '510300.SH'){
        tb_ctp_pre_price = select SecurityID,Price,tradetime,receivedTime  from csv_priceData where ID_type==`IF
    } elif(temp['SecurityID'][0] == '510050.SH'){
        tb_ctp_pre_price = select SecurityID,Price,tradetime,receivedTime  from csv_priceData where ID_type==`IH
    } elif(temp['SecurityID'][0] == '510500.SH'){
        tb_ctp_pre_price = select SecurityID,Price,tradetime,receivedTime  from csv_priceData where ID_type==`IC
    } elif(temp['SecurityID'][0] == '512100.SH'){
        tb_ctp_pre_price = select SecurityID,Price,tradetime,receivedTime  from csv_priceData where ID_type==`IM
    }

    update tb_ctp_pre_price set tradetime=temp['tradetime'][0]
    update tb_ctp_pre_price set receivedTime=temp['receivedTime'][0]
    tmp_final = select SecurityID,Price,tradetime,receivedTime from tb_ctp_pre_price
    reorderColumns!(tmp_final, reorderedColNames)
    objByName(hqTbName).append!(tmp_final)
}

2. 增加空表判断，避免越界访问 #

在访问temp的行数据前，先判断temp是否为空：

temp = select securityCode as SecurityID,preClosePrice,lastPrice as Price,origTime as tradetime,receivedTime from msg where (not isNull(lastPrice)) and lastPrice != 0
// 空表直接返回，不执行后续逻辑
if temp.size() == 0 {
    return
}
update temp set chg_pct= (double(Price)/double(preClosePrice))-1 

3. 异常捕获保证处理流程不中断 #

用try-catch包裹核心处理逻辑，确保单次异常不会影响后续消息处理：

def handleCtp_ETF_Subs(mutable msg, hqTbName, reorderedColNames,csv_priceData) {
    try {
        temp = select securityCode as SecurityID,preClosePrice,lastPrice as Price,origTime as tradetime,receivedTime from msg where (not isNull(lastPrice)) and lastPrice != 0
        if temp.size() == 0 {
            return
        }
        update temp set chg_pct= (double(Price)/double(preClosePrice))-1 

        tb_ctp_pre_price = table(0:0, ['SecurityID', 'Price', 'tradetime', 'receivedTime'], [STRING, DOUBLE, DATETIME, DATETIME])
        if(temp['SecurityID'][0] == '510300.SH'){
            tb_ctp_pre_price = select SecurityID,Price,tradetime,receivedTime  from csv_priceData where ID_type==`IF
        } elif(temp['SecurityID'][0] == '510050.SH'){
            tb_ctp_pre_price = select SecurityID,Price,tradetime,receivedTime  from csv_priceData where ID_type==`IH
        } elif(temp['SecurityID'][0] == '510500.SH'){
            tb_ctp_pre_price = select SecurityID,Price,tradetime,receivedTime  from csv_priceData where ID_type==`IC
        } elif(temp['SecurityID'][0] == '512100.SH'){
            tb_ctp_pre_price = select SecurityID,Price,tradetime,receivedTime  from csv_priceData where ID_type==`IM
        }

        update tb_ctp_pre_price set tradetime=temp['tradetime'][0]
        update tb_ctp_pre_price set receivedTime=temp['receivedTime'][0]
        tmp_final = select SecurityID,Price,tradetime,receivedTime from tb_ctp_pre_price
        reorderColumns!(tmp_final, reorderedColNames)
        objByName(hqTbName).append!(tmp_final)
    } catch (e) {
        // 可选：记录异常日志，方便排查问题
        print("处理异常: " + e)
    }
}

4. 验证流表append的原子性 #

确认objByName(hqTbName)获取的是正确的流表对象，若使用多线程处理流表，需使用线程安全的流表接口或添加锁机制，避免并发append操作引发的数据丢失。

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