远程Redis Pub/Sub传输Base64帧时Flutter接收数据不完整求助

我实现的流程是：视频→Python应用转为帧→将Base64编码的帧发布到Redis→Flutter订阅Redis、解码并展示。本地Redis测试时一切正常，但使用远程Redis服务器时，Flutter订阅后接收到的是不完整的Base64数据，无法正常解码，报错信息如下：
flutter: Error decoding base64 string: FormatException: Invalid length, must be multiple of four (at character 10036) ...LHr0Jr7z+BX7UfwP/am0eW28H6ujXsEQ/tPQdQQC4gyOjJyHX/aHFen9drUnG0tE00uzW33HHVw
使用Python订阅应用测试则完全正常。

Python发布帧代码 #

import redis
import base64
import cv2


def publish_frames(video_path, channel_name, redis_host='non-localhost-ip', redis_port=6380):
    # Connect to Redis server
    client = redis.StrictRedis(host=redis_host, port=redis_port, decode_responses=True, db=0)

    # Open the video file
    cap = cv2.VideoCapture(video_path)

    if not cap.isOpened():
        print(f"Error: Could not open video file {video_path}")
        return

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        try:
            print("Frames")
            # Encode the frame as a JPEG image
            _, buffer = cv2.imencode('.jpg', frame)

            # Convert the image buffer to a base64 string
            frame_base64 = base64.b64encode(buffer).decode('utf-8')

            # Ensure proper base64 padding (this should not be necessary in Python, but it's good to ensure)
            # if len(frame_base64) % 4 != 0:
            #     frame_base64 += '=' * (4 - len(frame_base64) % 4)

            # Publish the base64 string to the Redis channel
            client.publish(channel_name, frame_base64)
            cv2.waitKey(1)  # Adjust this delay as needed (e.g., 1 ms for high frame rate)
        except Exception as e:
            print(f"Error encoding or publishing frame: {e}")

        # Delay to simulate the video frame rate


    cap.release()
    print(f"Finished publishing frames from {video_path}")


if __name__ == '__main__':
    video_path = 'videoplayback.mp4'
    channel_name = 'video_frames'
    publish_frames(video_path, channel_name)

Flutter接收帧代码 #

import 'dart:convert';
import 'dart:typed_data';

import 'package:flutter/material.dart';
import 'package:ioredis/ioredis.dart';

void main() async {
  final redis = Redis(RedisOptions(host: 'non-localhost-ip', port: 6380));
  print('Connected to Redis');

  runApp(MyApp(redis: redis));
}

class MyApp extends StatelessWidget {
  final Redis redis;

  const MyApp({Key? key, required this.redis}) : super(key: key);

  @override
  Widget build(BuildContext context) {
    return MaterialApp(
      title: 'Redis Video Stream',
      theme: ThemeData(
        primarySwatch: Colors.blue,
      ),
      home: VideoStreamPage(redis: redis),
    );
  }
}

class VideoStreamPage extends StatefulWidget {
  final Redis redis;

  const VideoStreamPage({Key? key, required this.redis}) : super(key: key);

  @override
  _VideoStreamPageState createState() => _VideoStreamPageState();
}

class _VideoStreamPageState extends State<VideoStreamPage> {
  List<Uint8List> frames = [];

  @override
  void initState() {
    super.initState();
    startListening();
  }

  @override
  void dispose() {
    super.dispose();
    stopListening();
  }

  void startListening() async {
    try {
      print('Subscribing to video_frames...');
      RedisSubscriber subscriber = await widget.redis.subscribe('video_frames');
      print('Subscribed to video_frames');

      subscriber.onMessage = (String channel, String? message) async {
        if (message != null) {
          print('Received message: ${message.substring(0, 30)}...');  // Print start of the base64 string
          try {
            Uint8List bytes = base64.decode(message);
            print('Frame decoded successfully.');
            setState(() {
              frames.add(bytes);
            });
          } catch (e) {
            print('Error decoding base64 string: $e');
          }
        }
      };
    } catch (e) {
      print('Error: $e');
    }
  }

  void stopListening() async {
    await widget.redis.disconnect();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(
        title: Text('Video Stream'),
      ),
      body: ListView.builder(
        itemCount: frames.length,
        itemBuilder: (context, index) {
          return Image.memory(
            frames[index],
            fit: BoxFit.cover,
          );
        },
      ),
    );
  }
}

解决建议 #

• 修复Base64 padding：取消Python代码中注释的padding补全逻辑，强制确保Base64字符串长度为4的倍数，避免传输中padding丢失：

  if len(frame_base64) % 4 != 0:
      frame_base64 += '=' * (4 - len(frame_base64) % 4)
  

• 处理Flutter端消息分包：ioredis可能在大消息传输时触发多次回调，导致单帧数据被拆分。可以尝试更换redis包，或者在Flutter端添加消息拼接逻辑（比如通过判断消息结尾是否为Base64 padding字符，或给每个帧添加唯一标识）。
• 减小单帧数据大小：降低JPEG编码质量，减少Base64字符串长度，避免触发网络或Redis的消息大小限制：

  _, buffer = cv2.imencode('.jpg', frame, [cv2.IMWRITE_JPEG_QUALITY, 70])
  

• Flutter端自动补全padding：在解码前先检查并补全Base64字符串的padding，即使收到不完整数据也能尝试修复：

  String processedMsg = message;
  int padLength = 4 - (processedMsg.length % 4);
  if (padLength != 4) {
    processedMsg += '=' * padLength;
  }
  Uint8List bytes = base64.decode(processedMsg);
  

• 检查远程Redis配置：确认远程Redis的client-output-buffer-limit pubsub参数是否设置合理，避免大消息被截断。

内容的提问来源于stack exchange，提问作者Sukhanpreet Singh Dhillon