04 - 消息格式

消息格式（Messages）

4.1 消息概述

RTMP 消息（Message）是协议中 完整语义单元，一个消息表示一条完整的控制指令或媒体数据。在块流层之上，多个块重组为一个消息。

应用层视角：
┌─────────────────────────────────────────────────────┐
│                  RTMP Message                        │
│  ┌────────────┐  ┌─────────────────────────────┐   │
│  │  Header    │  │           Body               │   │
│  │  (11 bytes)│  │      (variable length)       │   │
│  └────────────┘  └─────────────────────────────┘   │
└─────────────────────────────────────────────────────┘

传输层视角（经过块流拆分）：
┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐
│Chunk #1│ │Chunk #2│ │Chunk #3│ │Chunk #4│
│(头+128)│ │(1B+128)│ │(1B+128)│ │(1B+44) │
└────────┘ └────────┘ └────────┘ └────────┘
           ↑ fmt=3    ↑ fmt=3    ↑ fmt=3 (后续块)

4.2 消息头结构

每个消息的头部是 11 字节（在 fmt=0 块中完整呈现）：

 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
│                  timestamp (3 bytes, big-endian)               │
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
│                message length (3 bytes, big-endian)            │
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
│   message type id (1 byte)    │                                 │
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
│             message stream id (4 bytes, little-endian)         │
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

字段详解：
┌─────────────────┬──────────────────────────────────────────────────────────┐
│  字段           │  说明                                                    │
├─────────────────┼──────────────────────────────────────────────────────────┤
│ timestamp       │ 消息时间戳（毫秒），支持绝对值或增量值                    │
│                 │ 范围: 0 ~ 0xFFFFFF (约 4.6 小时)                         │
│                 │ 超过范围时使用 4 字节扩展时间戳                            │
├─────────────────┼──────────────────────────────────────────────────────────┤
│ message length  │ 消息体的长度（字节），不包含消息头                         │
│                 │ 范围: 0 ~ 16,777,215 (2^24 - 1)                          │
├─────────────────┼──────────────────────────────────────────────────────────┤
│ message type    │ 消息类型 ID，1 字节                                       │
│                 │ 决定了 Body 的数据格式和语义                               │
├─────────────────┼──────────────────────────────────────────────────────────┤
│ message stream  │ 消息流 ID，标识所属的逻辑流                               │
│                 │ 注意: 小端序（Little-Endian）                             │
└─────────────────┴──────────────────────────────────────────────────────────┘

4.3 消息类型总览

RTMP 定义了丰富的消息类型，分为以下几大类：

协议控制消息（Protocol Control Messages）

AMF 命令消息（Command Messages）

数据消息

媒体消息

其他消息

4.4 协议控制消息详解

4.4.1 Set Chunk Size (Type 1)

在块流章节中已介绍，这里补充实现细节：

import struct

def encode_set_chunk_size(size: int) -> bytes:
    """
    编码 Set Chunk Size 消息
    - 消息类型: 1
    - 块流 ID: 2
    - Body: 4 字节，最高位必须为 0
    """
    if size < 1 or size > 0x7FFFFFFF:
        raise ValueError(f"Invalid chunk size: {size}")

    body = struct.pack('>I', size & 0x7FFFFFFF)
    return body


def decode_set_chunk_size(body: bytes) -> int:
    """解码 Set Chunk Size 消息"""
    return struct.unpack('>I', body[:4])[0] & 0x7FFFFFFF

4.4.2 User Control Message (Type 4)

用户控制消息用于传输控制事件，Body 以 2 字节事件类型开头：

Body 结构：
┌──────────────────┬──────────────────────────────┐
│  event type (2B) │  event data (variable)       │
└──────────────────┴──────────────────────────────┘

事件类型：
┌───────────┬──────────────────┬──────────────────────────────┐
│  类型 ID  │  名称            │  说明                        │
├───────────┼──────────────────┼──────────────────────────────┤
│  0        │  Stream Begin    │  流开始播放                   │
│  1        │  Stream EOF      │  流结束                       │
│  2        │  Stream Dry      │  流无数据                     │
│  3        │  Set Buffer Length│ 设置缓冲区大小（播放端）      │
│  4        │  Stream Is Recorded│ 流是录制流                  │
│  6        │  Ping Request    │  Ping 请求                   │
│  7        │  Ping Response   │  Ping 响应                   │
│  26       │  SWF Verification│  SWF 验证请求                │
│  27       │  SWF Response    │  SWF 验证响应                │
│  31       │  Buffer Empty    │  缓冲区空                     │
│  32       │  Buffer Ready    │  缓冲区就绪                   │
└───────────┴──────────────────┴──────────────────────────────┘

class UserControlEvents:
    STREAM_BEGIN = 0
    STREAM_EOF = 1
    STREAM_DRY = 2
    SET_BUFFER_LENGTH = 3
    STREAM_IS_RECORDED = 4
    PING_REQUEST = 6
    PING_RESPONSE = 7
    SWF_VERIFICATION = 26
    SWF_RESPONSE = 27
    BUFFER_EMPTY = 31
    BUFFER_READY = 32


def encode_user_control_stream_begin(stream_id: int) -> bytes:
    """编码 Stream Begin 事件"""
    return struct.pack('>HI', UserControlEvents.STREAM_BEGIN, stream_id)


def encode_user_control_set_buffer_length(stream_id: int, buffer_ms: int) -> bytes:
    """编码 Set Buffer Length 事件"""
    return struct.pack('>HII',
                        UserControlEvents.SET_BUFFER_LENGTH,
                        stream_id,
                        buffer_ms)


def encode_user_control_ping_request(timestamp: int) -> bytes:
    """编码 Ping Request"""
    return struct.pack('>HI', UserControlEvents.PING_REQUEST, timestamp)


def encode_user_control_ping_response(timestamp: int) -> bytes:
    """编码 Ping Response"""
    return struct.pack('>HI', UserControlEvents.PING_RESPONSE, timestamp)


def decode_user_control(body: bytes) -> tuple:
    """解码 User Control 消息，返回 (event_type, event_data)"""
    event_type = struct.unpack('>H', body[:2])[0]
    event_data = body[2:]
    return event_type, event_data

4.4.3 Window Acknowledgement Size (Type 5)

通知对端每收到多少字节后发送一次 Acknowledgement。

Body: 4 bytes，大端序，表示确认窗口大小（字节）

典型值: 2500000 (2.5 MB)

def encode_window_ack_size(size: int) -> bytes:
    return struct.pack('>I', size)


def decode_window_ack_size(body: bytes) -> int:
    return struct.unpack('>I', body[:4])[0]

4.4.4 Set Peer Bandwidth (Type 6)

限制对端的发送带宽。

Body 结构：
┌──────────────────────┬────────────────┐
│  window size (4B)    │  limit (1B)    │
└──────────────────────┴────────────────┘

limit 类型:
┌──────┬────────────────┬──────────────────────────────────┐
│ 值   │ 名称           │ 说明                             │
├──────┼────────────────┼──────────────────────────────────┤
│ 0    │ Hard           │ 对端必须限制发送速率              │
│ 1    │ Soft           │ 对端可以自行决定是否限制          │
│ 2    │ Dynamic        │ 若前次为 Hard 则变为 Hard，否则 Soft│
└──────┴────────────────┴──────────────────────────────────┘

class BandwidthLimit:
    HARD = 0
    SOFT = 1
    DYNAMIC = 2


def encode_set_peer_bandwidth(window_size: int, limit: int) -> bytes:
    return struct.pack('>IB', window_size, limit)


def decode_set_peer_bandwidth(body: bytes) -> tuple:
    window_size = struct.unpack('>I', body[:4])[0]
    limit = body[4]
    return window_size, limit

4.5 协议控制消息收发时序

典型的 RTMP 连接建立后的协议控制消息交换：

    Client                                     Server
      │                                          │
      │═══ 握手完成 ══════════════════════════════│
      │                                          │
      │── Set Chunk Size (4096) ────────────────→│  消息类型 1
      │                                          │
      │── Window Ack Size (2500000) ────────────→│  消息类型 5
      │                                          │
      │── Set Peer Bandwidth (2500000, Hard) ───→│  消息类型 6
      │                                          │
      │←── Set Chunk Size (4096) ───────────────│  消息类型 1
      │                                          │
      │←── Window Ack Size (2500000) ───────────│  消息类型 5
      │                                          │
      │←── Set Peer Bandwidth (2500000, Hard) ──│  消息类型 6
      │                                          │
      │←── User Control (Stream Begin, 0) ──────│  消息类型 4
      │                                          │
      │── connect("live") ─────────────────────→│  AMF 命令
      │                                          │
      │←── Window Ack Size (5000000) ───────────│
      │←── Set Peer Bandwidth (5000000, Hard) ──│
      │←── User Control (Stream Begin, 0) ──────│
      │←── _result (connect success) ───────────│  AMF 响应
      │                                          │

4.6 音频消息（Type 8）

音频消息承载音频编码数据，格式遵循 FLV Audio Tag 规范。

FLV 音频头

第一个字节（音频头）：
 0 1 2 3 4 5 6 7
+-+-+-+-+-+-+-+-+
│Sound  │Sound  │
│Format │Rate   │
│(4b)   │(2b)   │
+-+-+-+-+-+-+-+-+
│Sound  │Sound  │
│Size   │Type   │
│(1b)   │(1b)   │
+-+-+-+-+-+-+-+-+

┌──────────────┬────────┬──────────────────────────────┐
│  字段        │  位数  │  取值                         │
├──────────────┼────────┼──────────────────────────────┤
│ Sound Format │  4     │ 0=PCM, 1=ADPCM, 2=MP3,      │
│              │        │ 10=AAC, 11=Speex, ...        │
│ Sound Rate   │  2     │ 0=5.5kHz, 1=11kHz,          │
│              │        │ 2=22kHz, 3=44kHz             │
│ Sound Size   │  1     │ 0=8-bit, 1=16-bit           │
│ Sound Type   │  1     │ 0=mono, 1=stereo            │
└──────────────┴────────┴──────────────────────────────┘

AAC 音频消息格式

当 Sound Format = 10 (AAC) 时，Body 结构如下：

┌────────────────┬────────────────────────────────────┐
│  Audio Header  │         AAC Data                   │
│  (1 byte)      │                                    │
│  0xAF (典型)    │  ┌──────────┬──────────────────┐  │
│                 │  │AAC Packet│  AAC Raw Data    │  │
│                 │  │Type (1B) │  (ADTS/RAW)      │  │
│                 │  └──────────┴──────────────────┘  │
└────────────────┴────────────────────────────────────┘

AAC Packet Type:
  0 = AAC Sequence Header (解码器配置)
  1 = AAC Raw (实际音频帧)

def parse_audio_message(body: bytes) -> dict:
    """解析音频消息"""
    if len(body) < 1:
        return {}

    header = body[0]
    sound_format = (header >> 4) & 0x0F
    sound_rate = (header >> 2) & 0x03
    sound_size = (header >> 1) & 0x01
    sound_type = header & 0x01

    rate_map = {0: 5500, 1: 11000, 2: 22000, 3: 44100}

    result = {
        'sound_format': sound_format,
        'sound_format_name': {0: 'PCM', 1: 'ADPCM', 2: 'MP3',
                               10: 'AAC', 11: 'Speex'}.get(sound_format, 'Unknown'),
        'sound_rate': rate_map.get(sound_rate, 0),
        'sound_size': 8 if sound_size == 0 else 16,
        'sound_channels': 1 if sound_type == 0 else 2,
    }

    if sound_format == 10 and len(body) >= 2:
        # AAC
        result['aac_packet_type'] = body[1]
        result['aac_type'] = 'Sequence Header' if body[1] == 0 else 'Raw'

        if body[1] == 0:
            # 解析 AAC Sequence Header (AudioSpecificConfig)
            result['asc'] = parse_audio_specific_config(body[2:])

    return result


def parse_audio_specific_config(data: bytes) -> dict:
    """
    解析 AudioSpecificConfig (ISO 14496-3)
    最少 2 字节
    """
    if len(data) < 2:
        return {}

    audio_object_type = (data[0] >> 3) & 0x1F
    sampling_freq_index = ((data[0] & 0x07) << 1) | ((data[1] >> 7) & 0x01)
    channel_config = (data[1] >> 3) & 0x0F

    freq_map = {
        0: 96000, 1: 88200, 2: 64000, 3: 48000,
        4: 44100, 5: 32000, 6: 24000, 7: 22050,
        8: 16000, 9: 12000, 10: 11025, 11: 8000
    }

    return {
        'audio_object_type': audio_object_type,
        'sampling_freq_index': sampling_freq_index,
        'sampling_freq': freq_map.get(sampling_freq_index, 0),
        'channel_config': channel_config,
    }

4.7 视频消息（Type 9）

视频消息承载视频编码数据，格式遵循 FLV Video Tag 规范。

FLV 视频头

第一个字节（视频头）：
 0 1 2 3 4 5 6 7
+-+-+-+-+-+-+-+-+
│Frame  │Codec  │
│Type   │ID     │
│(4b)   │(4b)   │
+-+-+-+-+-+-+-+-+

┌──────────────┬────────┬──────────────────────────────┐
│  字段        │  位数  │  取值                         │
├──────────────┼────────┼──────────────────────────────┤
│ Frame Type   │  4     │ 1=Keyframe, 2=Inter,         │
│              │        │ 3=Disposable, 4=Generated     │
│ Codec ID     │  4     │ 2=Sorenson H.263, 7=AVC,     │
│              │        │ 12=HEVC (Enhanced RTMP)       │
└──────────────┴────────┴──────────────────────────────┘

H.264 视频消息格式

当 Codec ID = 7 (AVC/H.264) 时：

┌────────────┬───────────────┬──────────────────────────┐
│ Video Header│  AVC Packet   │       AVC Data           │
│  (1 byte)  │  Type (1B)    │                          │
│  0x17 (key) │               │  ┌────────────────────┐ │
│  0x27 (inter)│              │  │ Sequence Header:   │ │
│              │               │  │  SPS + PPS         │ │
│              │               │  │ NALU: 视频帧数据   │ │
│              │               │  └────────────────────┘ │
└──────────────┴───────────────┴──────────────────────────┘

AVC Packet Type:
  0 = AVC Sequence Header (SPS/PPS 解码器配置)
  1 = AVC NALU (实际视频帧)
  2 = AVC End of Sequence

AVC Sequence Header 结构 (AVCDecoderConfigurationRecord):
┌──────────────────┬────────────────────────────────────┐
│ version (1B)     │ 通常为 1                           │
│ profile (1B)     │ 如 100 = High                      │
│ compatibility (1B)│ 兼容性标志                        │
│ level (1B)       │ 如 40 = Level 4.0                 │
│ NALU length size │ 低 2 位 + 1 = NALU 长度前缀字节数  │
│ SPS count (1B)   │ SPS 数量（低 5 位）                │
│ SPS length (2B)  │ SPS 长度                           │
│ SPS data         │ SPS 数据                           │
│ PPS count (1B)   │ PPS 数量                           │
│ PPS length (2B)  │ PPS 长度                           │
│ PPS data         │ PPS 数据                           │
└──────────────────┴────────────────────────────────────┘

def parse_video_message(body: bytes) -> dict:
    """解析视频消息"""
    if len(body) < 1:
        return {}

    header = body[0]
    frame_type = (header >> 4) & 0x0F
    codec_id = header & 0x0F

    frame_type_names = {
        1: 'Keyframe', 2: 'Inter Frame',
        3: 'Disposable Inter', 4: 'Generated Keyframe'
    }
    codec_names = {
        2: 'Sorenson H.263', 3: 'Screen Video',
        4: 'VP6', 5: 'VP6 Alpha', 6: 'Screen Video v2',
        7: 'AVC (H.264)', 12: 'HEVC (H.265)'
    }

    result = {
        'frame_type': frame_type,
        'frame_type_name': frame_type_names.get(frame_type, 'Unknown'),
        'is_keyframe': frame_type == 1,
        'codec_id': codec_id,
        'codec_name': codec_names.get(codec_id, 'Unknown'),
    }

    if codec_id == 7 and len(body) >= 5:
        # AVC (H.264)
        result['avc_packet_type'] = body[1]
        composition_time = struct.unpack('>I', b'\x00' + body[2:5])[0]
        result['composition_time'] = composition_time

        if body[1] == 0:
            # AVC Sequence Header
            result['sequence_header'] = parse_avc_sequence_header(body[5:])
        elif body[1] == 1:
            # AVC NALU
            result['nalu'] = parse_avc_nalu(body[5:])

    return result


def parse_avc_sequence_header(data: bytes) -> dict:
    """解析 AVCDecoderConfigurationRecord"""
    if len(data) < 7:
        return {}

    config = {
        'version': data[0],
        'profile': data[1],
        'compatibility': data[2],
        'level': data[3],
        'nalu_length_size': (data[4] & 0x03) + 1,
        'sps': [],
        'pps': [],
    }

    # SPS
    sps_count = data[5] & 0x1F
    offset = 6
    for _ in range(sps_count):
        sps_len = struct.unpack('>H', data[offset:offset+2])[0]
        offset += 2
        config['sps'].append(data[offset:offset+sps_len])
        offset += sps_len

    # PPS
    pps_count = data[offset]
    offset += 1
    for _ in range(pps_count):
        pps_len = struct.unpack('>H', data[offset:offset+2])[0]
        offset += 2
        config['pps'].append(data[offset:offset+pps_len])
        offset += pps_len

    return config


def parse_avc_nalu(data: bytes, nalu_length_size: int = 4) -> list:
    """解析 AVC NALU 数据"""
    nalus = []
    offset = 0
    while offset < len(data):
        if offset + nalu_length_size > len(data):
            break
        nalu_len = int.from_bytes(data[offset:offset+nalu_length_size], 'big')
        offset += nalu_length_size
        if offset + nalu_len > len(data):
            break
        nalu_data = data[offset:offset+nalu_len]
        nalu_type = nalu_data[0] & 0x1F if nalu_data else 0
        nalus.append({
            'type': nalu_type,
            'type_name': {1: 'Slice', 5: 'IDR', 6: 'SEI',
                          7: 'SPS', 8: 'PPS'}.get(nalu_type, 'Other'),
            'size': nalu_len,
        })
        offset += nalu_len
    return nalus

4.8 Aggregate 消息（Type 22）

聚合消息将多个消息打包成一个，减少网络传输次数：

Aggregate 消息结构：
┌────────────────────────────────────────────────────┐
│  Sub-Message #1                                    │
│  ┌──────────────────────────────────────────┐      │
│  │  message type (1B)                        │      │
│  │  message length (3B)                      │      │
│  │  timestamp (3B)                           │      │
│  │  timestamp extended (1B)                  │      │
│  │  message stream id (3B)                   │      │
│  │  message data (variable)                  │      │
│  └──────────────────────────────────────────┘      │
│  Back Pointer #1 (4B): 指向前一个子消息大小         │
├────────────────────────────────────────────────────┤
│  Sub-Message #2                                    │
│  ┌──────────────────────────────────────────┐      │
│  │  ... 同上结构 ...                          │      │
│  └──────────────────────────────────────────┘      │
│  Back Pointer #2 (4B)                             │
└────────────────────────────────────────────────────┘

4.9 消息解码器完整实现

#!/usr/bin/env python3
"""
RTMP Message Decoder
完整的 RTMP 消息解析器
"""

import struct
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class RTMPMessage:
    """RTMP 消息"""
    timestamp: int
    msg_type_id: int
    msg_stream_id: int
    data: bytes
    msg_type_name: str = ''

    def __post_init__(self):
        type_names = {
            1: 'Set Chunk Size', 2: 'Abort', 3: 'Acknowledgement',
            4: 'User Control', 5: 'Window Ack Size',
            6: 'Set Peer Bandwidth', 8: 'Audio', 9: 'Video',
            15: 'AMF3 Data', 16: 'AMF3 Shared Object',
            17: 'AMF3 Command', 18: 'AMF0 Data',
            19: 'AMF0 Shared Object', 20: 'AMF0 Command',
            22: 'Aggregate'
        }
        self.msg_type_name = type_names.get(self.msg_type_id,
                                             f'Unknown({self.msg_type_id})')


class RTMPMessageDecoder:
    """RTMP 消息解码器"""

    def decode(self, msg: RTMPMessage) -> dict:
        """解码消息，返回结构化数据"""
        handlers = {
            1: self._decode_set_chunk_size,
            2: self._decode_abort,
            3: self._decode_acknowledgement,
            4: self._decode_user_control,
            5: self._decode_window_ack_size,
            6: self._decode_set_peer_bandwidth,
            8: self._decode_audio,
            9: self._decode_video,
        }

        handler = handlers.get(msg.msg_type_id)
        if handler:
            result = handler(msg.data)
        else:
            result = {'raw': msg.data.hex()}

        result['msg_type'] = msg.msg_type_name
        result['timestamp'] = msg.timestamp
        result['stream_id'] = msg.msg_stream_id
        return result

    def _decode_set_chunk_size(self, data: bytes) -> dict:
        size = struct.unpack('>I', data[:4])[0] & 0x7FFFFFFF
        return {'chunk_size': size}

    def _decode_abort(self, data: bytes) -> dict:
        cs_id = struct.unpack('>I', data[:4])[0]
        return {'abort_cs_id': cs_id}

    def _decode_acknowledgement(self, data: bytes) -> dict:
        seq = struct.unpack('>I', data[:4])[0]
        return {'sequence_number': seq}

    def _decode_user_control(self, data: bytes) -> dict:
        event_type = struct.unpack('>H', data[:2])[0]
        event_names = {
            0: 'Stream Begin', 1: 'Stream EOF', 2: 'Stream Dry',
            3: 'Set Buffer Length', 4: 'Stream Is Recorded',
            6: 'Ping Request', 7: 'Ping Response'
        }
        result = {
            'event_type': event_type,
            'event_name': event_names.get(event_type, 'Unknown')
        }
        if event_type in (0, 1, 2, 3, 4):
            result['stream_id'] = struct.unpack('>I', data[2:6])[0]
        if event_type == 3:
            result['buffer_length_ms'] = struct.unpack('>I', data[6:10])[0]
        if event_type in (6, 7):
            result['timestamp'] = struct.unpack('>I', data[2:6])[0]
        return result

    def _decode_window_ack_size(self, data: bytes) -> dict:
        return {'window_ack_size': struct.unpack('>I', data[:4])[0]}

    def _decode_set_peer_bandwidth(self, data: bytes) -> dict:
        return {
            'window_size': struct.unpack('>I', data[:4])[0],
            'limit_type': ['Hard', 'Soft', 'Dynamic'][data[4]]
        }

    def _decode_audio(self, data: bytes) -> dict:
        return parse_audio_message(data)

    def _decode_video(self, data: bytes) -> dict:
        return parse_video_message(data)

注意事项

1. 消息 vs 块：一个消息可能被拆成多个块，解码器必须先重组块再解析消息
2. 协议控制消息的块流 ID：类型 1-6 的消息必须使用块流 ID 2
3. 音视频消息的块流 ID：通常音频使用 8，视频使用 6（SRS 默认），但规范未强制
4. 时间戳精度：RTMP 时间戳单位是毫秒，注意与 PTS/DTS 的换算
5. 消息流 ID 字节序：消息流 ID 是小端序，与其他字段不同，容易混淆
6. Sequence Header 优先：音视频 Sequence Header（解码器配置）必须在数据帧之前发送

扩展阅读

• RTMP Specification - Message Format
• FLV File Format Specification
• ISO 14496-15 - AVC File Format

上一章：03 - 块流机制 下一章：05 - AMF 编码与命令 — 了解 RTMP 的命令通信机制