您现在的位置是：首页 > 文章详情

Monibuca v4.6.5 发布 —— 开源 Go 语言流媒体服务器

日期：2023-07-02点击：304收藏

发布摘要

启动工程增加对流的发布订阅的单元测试和基准测试
修复发布和订阅并发创建流产生的panic
修复IdleTimeout无效问题
修复订阅者阻塞导致读写并发问题
preview插件修复https默认端口
gb28181插件PR合并

升级模块升级后版本

engine 4.13.5

gb28181 4.3.6

hls 4.3.1

preview 4.1.2

record 4.3.7

详细说明

单元测试和基准测试

测试用例所在目录：github.com/langhuihui/monibuca工程下的test目录

推荐使用vscode打开工程，方便一键测试单个用例，也可以通过vscode侧边栏里的单元测试栏目管理。

TestPubAndSub

 func TestPubAndSub(t *testing.T) {  t.Cleanup(FreeEngine)  UseEngine()  t.Run("publish", func(t *testing.T) {   t.Parallel()   var pub UnitTestPublisher   unitTestPlugin.Publish("test/001", &pub)  })  t.Run("subscribe", func(t *testing.T) {   t.Parallel()   var sub UnitTestSubsciber   sub.TB = t   err := unitTestPlugin.Subscribe("test/001", &sub)   if err != nil {    t.Fatal(err)   } else {    sub.PlayRaw()   }  }) }

该用例会启动一个发布者和一个订阅者，发布者将会写入假数据，订阅者读取后会校验数据是否正确。

BenchmarkPubAndSub

 func BenchmarkPubAndSub(b *testing.B) {  b.Cleanup(FreeEngine)  UseEngine()  for i := 0; i < 10; i++ {   i := i   go func(i int) {    var pub UnitTestPublisher    unitTestPlugin.Publish(fmt.Sprintf("testb/%d", i), &pub)   }(i)   go b.RunParallel(func(pb *testing.PB) {    var sub UnitTestSubsciber    sub.TB = b    err := unitTestPlugin.Subscribe(fmt.Sprintf("testb/%d", i), &sub)    if err != nil {     // b.Fatal(err)    } else {     sub.PlayRaw()    }   })  }  time.Sleep(time.Second * 10) }

该用例为基准测试，启动10个发布者，每个流会批量订阅，以测试并发性能以及稳定性。

TestSlowSubscriber

 func TestSlowSubscriber(t *testing.T) {  t.Cleanup(FreeEngine)  UseEngine()  var pub UnitTestPublisher  unitTestPlugin.Publish("test/slow", &pub)  var suber SlowSubsciber  unitTestPlugin.Subscribe("test/slow", &suber)  suber.PlayRaw() }

这个用例会模拟一个订阅者被阻塞（sleep）后导致发布者写入的位置追上，此时engine通过标记这个写入点废弃来避免并发读写问题，废弃后订阅者如果被唤醒就会自动停止订阅。

通常出现这种情况是订阅者进行了某种耗时操作，比如写文件，或者网络阻塞等。

其他修复问题

并发创建流

修复如下：

 func findOrCreateStream(streamPath string, waitTimeout time.Duration) (s *Stream, created bool) {  p := strings.Split(streamPath, "/")  if len(p) < 2 {   log.Warn(Red("Stream Path Format Error:"), streamPath)   return nil, false  }  actual, loaded := Streams.LoadOrStore(streamPath, &Stream{   Path:       streamPath,   AppName:    p[0],   StreamName: strings.Join(p[1:], "/"),   StartTime:  time.Now(), +  Logger:     log.LocaleLogger.With(zap.String("stream", streamPath)), +  timeout:    time.NewTimer(waitTimeout),  })  if s := actual.(*Stream); loaded {   s.Debug("Stream Found")   return s, false  } else { -  s.timeout = time.NewTimer(waitTimeout)   s.Subscribers.Init() -  s.Logger = log.LocaleLogger.With(zap.String("stream", streamPath)) -  s.Info("created")   s.actionChan.Init(1) +  s.Info("created")   go s.run()   return s, true  } }

原理：当并发调用findOrCreateStream函数时，前者尚未来得及对Logger赋值，后者就调用了s.Debug导致空指针错误。