lq/nl-im-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复了一些问题

Browse Source
This commit is contained in:
李琦
2026-08-24 15:29:53 +08:00
parent 3417856607
commit cb15328e5d
47 changed files with 5514 additions and 966 deletions
Download Patch File Download Diff File Expand all files Collapse all files

156
internal/mediaserver/ffmpeg_transcoder.go
View File

@@ -21,12 +21,12 @@ import (
type TranscoderConfig struct {
// 模式配置
CopyMode bool // 是否使用 copy 模式（H.264 输入时使用，只重封装不重新编码）
// 视频配置
VideoCodec string // 输出视频编解码器，默认 libx264
VideoPreset string // x264 预设，默认 ultrafast
VideoBitrate string // 视频码率，默认 500k
// 音频配置
AudioCodec string // 输出音频编解码器，默认 aac
AudioBitrate string // 音频码率，默认 64k
@@ -51,18 +51,18 @@ func DefaultTranscoderConfig() *TranscoderConfig {
type FFmpegTranscoder struct {
config *TranscoderConfig
ffmpegPath string
cmd *exec.Cmd
stdin io.WriteCloser
stdout io.ReadCloser
stderr io.ReadCloser
outputChan chan []byte
stopChan chan struct{}
mu sync.Mutex
running bool
mu sync.Mutex
running bool
// 统计信息
inputBytes int64
outputBytes int64
@@ -74,13 +74,13 @@ func NewFFmpegTranscoder(config *TranscoderConfig) (*FFmpegTranscoder, error) {
if config == nil {
config = DefaultTranscoderConfig()
}
// 确保 FFmpeg 可用
ffmpegPath, err := EnsureFFmpeg()
if err != nil {
return nil, fmt.Errorf("FFmpeg 不可用: %w", err)
}
return &FFmpegTranscoder{
config: config,
ffmpegPath: ffmpegPath,
@@ -93,35 +93,35 @@ func NewFFmpegTranscoder(config *TranscoderConfig) (*FFmpegTranscoder, error) {
func (t *FFmpegTranscoder) Start() error {
t.mu.Lock()
defer t.mu.Unlock()
if t.running {
return fmt.Errorf("转码器已在运行")
}
// 构建 FFmpeg 命令
var args []string
if t.config.CopyMode {
// H.264 copy 模式：只重封装，不重新编码（低延迟）
// 添加低延迟参数：禁用缓冲、减少探测时间
args = []string{
// 低延迟输入参数
"-fflags", "nobuffer", // 禁用输入缓冲
"-flags", "low_delay", // 低延迟模式
"-probesize", "32", // 减少探测大小（字节）
"-analyzeduration", "0", // 禁用分析时长
"-f", "webm", // 输入格式
"-i", "pipe:0", // 从 stdin 读取
"-fflags", "nobuffer", // 禁用输入缓冲
"-flags", "low_delay", // 低延迟模式
"-probesize", "32", // 减少探测大小（字节）
"-analyzeduration", "0", // 禁用分析时长
"-f", "webm", // 输入格式
"-i", "pipe:0", // 从 stdin 读取
// 输出参数
"-c:v", "copy", // 视频直接复制，不重新编码
"-c:v", "copy", // 视频直接复制，不重新编码
"-c:a", t.config.AudioCodec,
"-b:a", t.config.AudioBitrate,
"-ar", fmt.Sprintf("%d", t.config.AudioSampleRate),
"-ac", fmt.Sprintf("%d", t.config.AudioChannels),
// 低延迟输出参数
"-fflags", "+genpts", // 生成时间戳
"-f", "flv", // 输出格式
"pipe:1", // 输出到 stdout
"-fflags", "+genpts", // 生成时间戳
"-f", "flv", // 输出格式
"pipe:1", // 输出到 stdout
}
log.Printf("🎬 [FFmpegTranscoder] 使用 copy 模式（H.264 重封装，低延迟）")
} else {
@@ -129,106 +129,109 @@ func (t *FFmpegTranscoder) Start() error {
// 添加低延迟参数
args = []string{
// 低延迟输入参数
"-fflags", "nobuffer", // 禁用输入缓冲
"-flags", "low_delay", // 低延迟模式
"-probesize", "32", // 减少探测大小（字节）
"-analyzeduration", "0", // 禁用分析时长
"-f", "webm", // 输入格式
"-i", "pipe:0", // 从 stdin 读取
"-fflags", "nobuffer", // 禁用输入缓冲
"-flags", "low_delay", // 低延迟模式
"-probesize", "32", // 减少探测大小（字节）
"-analyzeduration", "0", // 禁用分析时长
"-f", "webm", // 输入格式
"-i", "pipe:0", // 从 stdin 读取
// 视频编码参数
"-c:v", t.config.VideoCodec,
"-preset", t.config.VideoPreset,
"-tune", "zerolatency", // 零延迟调优
"-tune", "zerolatency", // 零延迟调优
"-b:v", t.config.VideoBitrate,
"-g", "30", // GOP 大小（减少关键帧间隔）
"-keyint_min", "15", // 最小关键帧间隔
"-g", "30", // GOP 大小（减少关键帧间隔）
"-keyint_min", "15", // 最小关键帧间隔
// 音频编码参数
"-c:a", t.config.AudioCodec,
"-b:a", t.config.AudioBitrate,
"-ar", fmt.Sprintf("%d", t.config.AudioSampleRate),
"-ac", fmt.Sprintf("%d", t.config.AudioChannels),
// 低延迟输出参数
"-fflags", "+genpts", // 生成时间戳
"-f", "flv", // 输出格式
"pipe:1", // 输出到 stdout
"-fflags", "+genpts", // 生成时间戳
"-f", "flv", // 输出格式
"pipe:1", // 输出到 stdout
}
log.Printf("🎬 [FFmpegTranscoder] 使用转码模式（VP8/VP9 → H.264，低延迟）")
}
t.cmd = exec.Command(t.ffmpegPath, args...)
var err error
// 获取 stdin
t.stdin, err = t.cmd.StdinPipe()
if err != nil {
return fmt.Errorf("获取 stdin 失败: %w", err)
}
// 获取 stdout
t.stdout, err = t.cmd.StdoutPipe()
if err != nil {
return fmt.Errorf("获取 stdout 失败: %w", err)
}
// 获取 stderr
t.stderr, err = t.cmd.StderrPipe()
if err != nil {
return fmt.Errorf("获取 stderr 失败: %w", err)
}
// 启动 FFmpeg 进程
if err := t.cmd.Start(); err != nil {
return fmt.Errorf("启动 FFmpeg 失败: %w", err)
}
t.running = true
t.startTime = time.Now()
log.Printf("🎬 [FFmpegTranscoder] 转码器已启动, PID=%d", t.cmd.Process.Pid)
// 启动输出读取协程
go t.readOutput()
// 启动 stderr 读取协程（用于日志）
go t.readStderr()
// 启动进程监控协程
go t.monitor()
return nil
}
// Stop 停止转码器
// 注意两个并发安全约束：
// 1. 不在此处调用 cmd.Wait()——monitor 协程是唯一的 Wait 回收方，
// os/exec 不允许对同一 Cmd 重复 Wait（第二次行为未定义）；
// 2. 不在此处 close(outputChan)——readOutput 协程（唯一发送方）可能仍在
// 向通道发送数据，向已关闭通道发送会 panic 并直接崩溃整个进程；
// 通道统一由 readOutput 退出时关闭（"由发送方关闭"的 Go 惯例）。
func (t *FFmpegTranscoder) Stop() error {
t.mu.Lock()
defer t.mu.Unlock()
if !t.running {
return nil
}
t.running = false
close(t.stopChan)
// 关闭 stdin，通知 FFmpeg 输入结束
if t.stdin != nil {
t.stdin.Close()
}
// 等待进程结束
// 杀掉进程即可：stdout 管道随之 EOF，readOutput 退出并关闭 outputChan；
// 僵尸进程由 monitor 协程的 cmd.Wait() 统一回收
if t.cmd != nil && t.cmd.Process != nil {
t.cmd.Process.Kill()
t.cmd.Wait()
}
// 关闭输出通道
close(t.outputChan)
duration := time.Since(t.startTime)
log.Printf("🛑 [FFmpegTranscoder] 转码器已停止, 运行时间=%v, 输入=%d bytes, 输出=%d bytes",
duration, t.inputBytes, t.outputBytes)
return nil
}
@@ -241,12 +244,12 @@ func (t *FFmpegTranscoder) Write(data []byte) (int, error) {
}
stdin := t.stdin
t.mu.Unlock()
n, err := stdin.Write(data)
if err != nil {
return n, err
}
t.inputBytes += int64(n)
return n, nil
}
@@ -257,16 +260,19 @@ func (t *FFmpegTranscoder) Output() <-chan []byte {
}
// readOutput 读取 FFmpeg 输出
// 本协程是 outputChan 的唯一发送方，因此由它在退出时关闭通道：
// 既保证 range Output() 的消费者能正常结束，又杜绝"向已关闭通道发送"的 panic
func (t *FFmpegTranscoder) readOutput() {
defer close(t.outputChan)
buf := make([]byte, 64*1024) // 64KB 缓冲区
for {
select {
case <-t.stopChan:
return
default:
}
n, err := t.stdout.Read(buf)
if err != nil {
if err != io.EOF {
@@ -274,14 +280,14 @@ func (t *FFmpegTranscoder) readOutput() {
}
return
}
if n > 0 {
t.outputBytes += int64(n)
// 复制数据并发送到通道
data := make([]byte, n)
copy(data, buf[:n])
select {
case t.outputChan <- data:
default:
@@ -296,7 +302,7 @@ func (t *FFmpegTranscoder) readOutput() {
func (t *FFmpegTranscoder) readStderr() {
buf := new(bytes.Buffer)
io.Copy(buf, t.stderr)
if buf.Len() > 0 {
// 只在有错误时输出
output := buf.String()
@@ -310,12 +316,12 @@ func (t *FFmpegTranscoder) readStderr() {
// monitor 监控 FFmpeg 进程
func (t *FFmpegTranscoder) monitor() {
err := t.cmd.Wait()
t.mu.Lock()
wasRunning := t.running
t.running = false
t.mu.Unlock()
if wasRunning {
if err != nil {
log.Printf("⚠️ [FFmpegTranscoder] FFmpeg 进程异常退出: %v", err)
@@ -361,21 +367,21 @@ func NewTranscoderPool(config *TranscoderConfig) *TranscoderPool {
func (p *TranscoderPool) Get(streamID string) (*FFmpegTranscoder, error) {
p.mu.Lock()
defer p.mu.Unlock()
if t, exists := p.transcoders[streamID]; exists && t.IsRunning() {
return t, nil
}
// 创建新的转码器
t, err := NewFFmpegTranscoder(p.config)
if err != nil {
return nil, err
}
if err := t.Start(); err != nil {
return nil, err
}
p.transcoders[streamID] = t
return t, nil
}
@@ -384,7 +390,7 @@ func (p *TranscoderPool) Get(streamID string) (*FFmpegTranscoder, error) {
func (p *TranscoderPool) Release(streamID string) {
p.mu.Lock()
defer p.mu.Unlock()
if t, exists := p.transcoders[streamID]; exists {
t.Stop()
delete(p.transcoders, streamID)
@@ -395,7 +401,7 @@ func (p *TranscoderPool) Release(streamID string) {
func (p *TranscoderPool) ReleaseAll() {
p.mu.Lock()
defer p.mu.Unlock()
for id, t := range p.transcoders {
t.Stop()
delete(p.transcoders, id)
@@ -408,5 +414,3 @@ func (p *TranscoderPool) Count() int {
defer p.mu.RUnlock()
return len(p.transcoders)
}