React Native接收MQTT高频数据存入Redux致应用卡顿求助

通过react-native-mqtt包从MQTT服务器接收传感器数据，每秒约250条。需逐条校验后存入Redux，但当前应用因数据处理与Redux操作出现严重卡顿，现有优化手段无效，求有效解决方案。

核心处理代码 #

var timestamp = []
var result = true

const onMessageArrived = async (theMessage, dispatch) => { 
  const split = theMessage.payloadString.split(';')
  if (timestamp.length > 0) {
    result = JSON.stringify(split) != JSON.stringify(timestamp)
  }

  if (result) {
    if (split[0] == 'bms100') {
      'worklet';
      BMS100(split, dispatch)
      dispatch({
          type: CHANGE_BMS100, payload: true
      })
    }
    else if (split[0] == 'bms24') {
      'worklet';
      BMS24(split, dispatch)
      dispatch({
          type: CHANGE_BMS24, payload: true
      })
    }
    else if (split[0] == 'n2k') {
      'worklet';
      N2K(split, dispatch)
      dispatch({
          type: CHANGE_N2K, payload: true
      })
    }
    else if (split[0] == 'motor') {
      'worklet';
      Motor(split, dispatch)
      dispatch({
          type: CHANGE_MOTOR, payload: true
      })
    }
    else if (split[0] == 'modbus') {
      'worklet';
      Modus(split, dispatch)
      dispatch({
          type: CHANGE_MODBUS, payload: true
      })
    } else {
      dispatch({
          type: FALSE_HEART_BEAT, payload: true
      })
    }
  }

  timestamp = split
}

Modus函数代码 #

export const Modus = (split, dispatch) => {
  if (split[0] == 'modbus' && split[1] == 'lower_deck_bb') {
    if (
      store.getState().LowerDeckLeftBtn.washingMachine != booleanRev(split[3])
    ) {
      dispatch({
        type: SYSTEM_MAIN_LEFT_WASHING_MACHINE,
        payload: booleanRev(split[3]),
      });
    }
    if (
      store.getState().LowerDeckLeftBtn.grayWaterPump != booleanRev(split[4])
    ) {
      dispatch({
        type: SYSTEM_MAIN_LEFT_GRAY_WATER_PUMP,
        payload: booleanRev(split[4]),
      });
    }
    if (store.getState().LowerDeckLeftBtn.toilet != booleanRev(split[5])) {
      dispatch({
        type: SYSTEM_MAIN_LEFT_TOLIT,
        payload: booleanRev(split[5]),
      });
    }

    if (store.getState().LowerDeckLeftBtn.ventilation != booleanRev(split[6])) {
      dispatch({
        type: SYSTEM_MAIN_LEFT_VENTILATION,
        payload: booleanRev(split[6]),
      });
    }
    if (
      store.getState().LowerDeckLeftBtn.bilgepumpcenter != booleanRev(split[8])
    ) {
      dispatch({
        type: SYSTEM_MAIN_LEF_BILGE_PUMP_CENTER,
        payload: booleanRev(split[8]),
      });
    }
    if (store.getState().LowerDeckLeftBtn.boiler != booleanRev(split[9])) {
      dispatch({
        type: SYSTEM_MAIN_LEFT_BOILER,
        payload: booleanRev(split[9]),
      });
    }
    if (store.getState().LowerDeckLeftBtn.aircon != booleanRev(split[10])) {
      dispatch({
        type: SYSTEM_MAIN_LEFT_AIR_CON,
        payload: booleanRev(split[10]),
      });
    }
    if (store.getState().LowerDeckLeftBtn.waterpump != booleanRev(split[7])) {
      dispatch({
        type: SYSTEM_MAIN_LEFT_WATER_PUMP,
        payload: booleanRev(split[7]),
      });
    }
    if (
      store.getState().LowerDeckLeftBtn.bilgepumpaft != booleanRev(split[11])
    ) {
      dispatch({
        type: SYSTEM_MAIN_LEFT_BLIGE_AFT,
        payload: booleanRev(split[11]),
      });
    }

    if (
      store.getState().LowerDeckRightBtn.ventilation != booleanRev(split[6])
    ) {
      dispatch({
        type: LOWERDECK_RIGHT_VENTILATION,
        payload: booleanRev(split[6]),
      });
    }
    if (store.getState().LowerDeckRightBtn.aircon != booleanRev(split[10])) {
      dispatch({
        type: LOWERDECK_RIGHT_AIRCON,
        payload: booleanRev(split[10]),
      });
    }
    if (store.getState().LowerDeckRightBtn.toilet != booleanRev(split[5])) {
      dispatch({
        type: LOWERDECK_RIGHT_TOLIT,
        payload: booleanRev(split[5]),
      });
    }
    if (
      store.getState().LowerDeckRightBtn.grayWaterPump != booleanRev(split[4])
    ) {
      dispatch({
        type: LOWERDECK_RIGHT_GRAY_WATER_PUMP,
        payload: booleanRev(split[4]),
      });
    }
    if (
      store.getState().LowerDeckRightBtn.bilgepumpaft != booleanRev(split[11])
    ) {
      dispatch({
        type: LOWERDECK_RIGHT_BILGE_PUMP_AFT,
        payload: booleanRev(split[11]),
      });
    }
    if (
      store.getState().LowerDeckRightBtn.bilgepumpcenter != booleanRev(split[8])
    ) {
      dispatch({
        type: LOWERDECK_RIGHT_BILGE_PUMP_CENTER,
        payload: booleanRev(split[8]),
      });
    }
  }

  else if (split[0] == 'modbus' && split[1] == 'switches') {
    if (store.getState().LowerDeckSwitch.lightSBB != booleanRev(split[3])) {
      dispatch({
        type: LOWER_DECK_LIGHT_BB,
        payload: booleanRev(split[3]),
      });
    }
    if (store.getState().LowerDeckSwitch.lightSTB != booleanRev(split[4])) {
      dispatch({
        type: LOWER_DECK_LIGHT_STB,
        payload: booleanRev(split[4]),
      });
    }
    if (store.getState().LowerDeckSwitch.outletsBB != booleanRev(split[5])) {
      dispatch({
        type: LOWER_DECK_OUTLETS_BB,
        payload: booleanRev(split[5]),
      });
    }

    if (store.getState().LowerDeckSwitch.outletsstb != booleanRev(split[6])) {
      dispatch({
        type: LOWER_DECK_OUTLETS_STB,
        payload: booleanRev(split[6]),
      });
    }

    if (store.getState().UpperDeckSwitch.lightsalon != booleanRev(split[7])) {
      dispatch({
        type: UPPER_DECK_LIGHT_SALON,
        payload: booleanRev(split[7]),
      });
    }
    if (store.getState().UpperDeckSwitch.outletsalon != booleanRev(split[8])) {
      dispatch({
        type: UPPER_DECK_OUTLETS_SALON,
        payload: booleanRev(split[8]),
      });
    }
    if (store.getState().UpperDeckSwitch.lightOutside != booleanRev(split[9])) {
      dispatch({
        type: UPPER_DECK_LIGHT_OUTSIDE,
        payload: booleanRev(split[9]),
      });
    }
    if (
      store.getState().UpperDeckSwitch.navEquipment != booleanRev(split[10])
    ) {
      dispatch({
        type: UPPER_DECK_NAV_EQUIPMENT,
        payload: booleanRev(split[10]),
      });
    }
  } else if (split[0] == 'modbus' && split[1] == 'status') {
    if (store.getState().WaterDieselEvent.freshWaterbb != split[5]) {
      dispatch({
        type: CHANGE_FRESHWATER_BB,
        payload: split[5],
      });
    }
    if (store.getState().WaterDieselEvent.freshwaterstb != split[4]) {
      dispatch({
        type: CHANGE_FRESHWATER_STB,
        payload: split[4],
      });
    }
    if (store.getState().WaterDieselEvent.blackwaterbb != split[7]) {
      dispatch({
        type: CHANGE_BLACKWATER_BB,
        payload: split[7],
      });
    }
    if (store.getState().WaterDieselEvent.blackwaterstb != split[6]) {
      dispatch({
        type: CHANGE_BLACKWATER_STB,
        payload: split[6],
      });
    }
    if (store.getState().WaterDieselEvent.diesel != split[8]) {
      dispatch({
        type: CHANGE_DIESEL,
        payload: split[8],
      });
    }
    if (
      store.getState().GeneratorAndPower.GeneratorSwitch != booleanRev(split[3])
    ) {
      dispatch({
        type: CHANGE_GENERATOR_SWITCH,
        payload: booleanRev(split[3]),
      });
    }
  }
};

优化解决方案 #

1. 批量处理数据，减少Redux更新频率 #

每秒250条数据逐条处理会触发大量Redux dispatch，阻塞JS线程。改用队列缓存数据，每隔固定时间批量处理：

const createMessageProcessor = () => {
  let messageQueue = [];
  let isProcessing = false;
  const BATCH_INTERVAL = 100; // 每100ms处理一次

  const processQueue = (dispatch) => {
    if (messageQueue.length === 0) {
      isProcessing = false;
      return;
    }
    const messages = [...messageQueue];
    messageQueue = [];

    messages.forEach(msg => {
      const split = msg.payloadString.split(';');
      handleSingleMessage(split, dispatch);
    });

    setTimeout(() => processQueue(dispatch), BATCH_INTERVAL);
  };

  return (theMessage, dispatch) => {
    messageQueue.push(theMessage);
    if (!isProcessing) {
      isProcessing = true;
      setTimeout(() => processQueue(dispatch), BATCH_INTERVAL);
    }
  };
};

// 替换原onMessageArrived
const onMessageArrived = createMessageProcessor();

// 抽离原处理逻辑
const handleSingleMessage = (split, dispatch) => {
  // 保留原校验、分支处理逻辑
};

2. 替换低效的数组比较逻辑 #

JSON.stringify对比数组性能极差，改用逐元素比较：

const arraysAreEqual = (arr1, arr2) => {
  if (arr1.length !== arr2.length) return false;
  for (let i = 0; i < arr1.length; i++) {
    if (arr1[i] !== arr2[i]) return false;
  }
  return true;
};

// 替换原比较逻辑
result = !arraysAreEqual(split, timestamp);

3. 缓存重复计算与Store状态 #

Modus函数中多次调用store.getState()和booleanRev，提前缓存减少重复操作：

export const Modus = (split, dispatch) => {
  // 提前计算反转后的布尔值
  const revMap = {
    3: booleanRev(split[3]),
    4: booleanRev(split[4]),
    5: booleanRev(split[5]),
    6: booleanRev(split[6]),
    7: booleanRev(split[7]),
    8: booleanRev(split[8]),
    9: booleanRev(split[9]),
    10: booleanRev(split[10]),
    11: booleanRev(split[11])
  };

  // 一次性获取Store状态
  const state = store.getState();
  const leftBtn = state.LowerDeckLeftBtn;
  const rightBtn = state.LowerDeckRightBtn;

  if (split[0] == 'modbus' && split[1] == 'lower_deck_bb') {
    if (leftBtn.washingMachine !== revMap[3]) {
      dispatch({ type: SYSTEM_MAIN_LEFT_WASHING_MACHINE, payload: revMap[3] });
    }
    if (leftBtn.grayWaterPump !== revMap[4]) {
      dispatch({ type: SYSTEM_MAIN_LEFT_GRAY_WATER_PUMP, payload: revMap[4] });
    }
    // 其余判断逻辑复用缓存变量
  }
};

4. 合并多个Dispatch为批量更新 #

创建批量更新Action，在Reducer中一次性处理所有变更，减少Redux执行次数：

// 定义批量更新Action类型
const BATCH_UPDATE_MODBUS = 'BATCH_UPDATE_MODBUS';

// 收集更新后批量Dispatch
export const Modus = (split, dispatch) => {
  const updates = {};
  const state = store.getState();
  const rev3 = booleanRev(split[3]);

  if (split[0] == 'modbus' && split[1] == 'lower_deck_bb') {
    if (state.LowerDeckLeftBtn.washingMachine !== rev3) {
      updates.washingMachine = rev3;
    }
    // 收集所有需要更新的字段
  }

  if (Object.keys(updates).length > 0) {
    dispatch({ type: BATCH_UPDATE_MODBUS, payload: { type: split[1], updates } });
  }
};

// Reducer中处理批量更新
const rootReducer = (state, action) => {
  switch (action.type) {
    case BATCH_UPDATE_MODBUS:
      const { type, updates } = action.payload;
      if (type === 'lower_deck_bb') {
        return {
          ...state,
          LowerDeckLeftBtn: { ...state.LowerDeckLeftBtn, ...updates },
          LowerDeckRightBtn: { ...state.LowerDeckRightBtn, ...updates }
        };
      }
      // 其他类型的批量更新逻辑
    default:
      return state;
  }
};

5. 正确使用Worklet处理数据 #

原代码中'worklet';仅为注释，未实际启用Worklet。使用Reanimated将数据处理移至非UI线程：

import { runOnUI, runOnJS } from 'react-native-reanimated';

const onMessageArrived = (theMessage, dispatch) => {
  runOnUI(() => {
    const split = theMessage.payloadString.split(';');
    const isNew = !arraysAreEqual(split, timestamp);
    if (isNew) {
      runOnJS(handleSingleMessage)(split, dispatch);
    }
    timestamp = split;
  })();
};

6. 优化MQTT接收配置 #

• 调整QoS级别：若无需每条数据可靠到达，将QoS设为0，减少客户端与服务器交互开销。
• 服务器端合并数据：要求服务器每隔一段时间合并同类型传感器数据，降低每秒发送条数。

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