FFmpeg mit Scala: Echtzeitvideo
Starten Sie mit Scala einen RTSP-Live-Testfeed, transformieren Sie ihn mit FFmpeg und liefern Sie den resultierenden HLS-Stream über lokales HTTP aus. Sie sehen ein bewegtes Graustufenmuster mit Zeitstempel und behalten eine 12 Sekunden lange Aufnahme. Das Beispiel läuft unter Linux, benötigt keine Kamera und beendet seine eigenen Dienste am Ende der Demonstration.
Entscheidend ist die Kontrolle über die eigenen Prozesse: Scala startet FFmpeg und wartet auf
dessen tatsächliches Ende. Ein Zeitlimit und ein Shutdown-Hook können den Prozess dabei stoppen.
Allein das Warten auf ein Scala-Objekt vom Typ Future bricht einen externen
Encoder nicht ab. Dies ist ein einzelner Stream ohne Audio in einer Qualitätsstufe mit üblicher
HLS-Pufferung. Er demonstriert weder eine Ziellatenz für den Produktivbetrieb noch eine Auslieferung
mit adaptiver Bitrate.
Scala und FFmpeg einrichten
Diese Anleitung wurde unter Linux x86-64 mit OpenJDK 21.0.12.1, Scala CLI 1.17.1,
Scala 2.13.18, MediaMTX 1.21.1 und
FFmpeg/ffprobe/ffplay 9.0.1 getestet. Verwenden Sie Bash für die Befehle. Installieren Sie einen
FFmpeg-Build für Linux mit libx264, dem
HLS-Muxer und den Filtern testsrc2, scale, hue, drawbox und drawtext.
Sie benötigen außerdem das genannte JDK, cURL, gzip, tar, awk, das fontconfig-Werkzeug
fc-match und eine installierte TrueType-Schrift wie Liberation Sans.
FFplay benötigt einen grafischen Desktop für die sichtbare Wiedergabe.
java -version && ffmpeg -version && ffprobe -version && ffplay -version &&
fc-match 'Liberation Sans'
Erstellen Sie ein neues Projekt und laden Sie die festgelegten Versionen von
Scala CLI und
MediaMTX herunter. MediaMTX ist die lokale RTSP-Gegenstelle:
FFmpeg veröffentlicht dort einen Stream, den anschließend ein weiterer FFmpeg-Prozess liest.
Fügen Sie diesen Block in einem beschreibbaren übergeordneten Verzeichnis ein. Wenn
scala-live-demo bereits existiert, stoppt die Einrichtung, ohne dieses Verzeichnis
zu ändern. Wählen Sie für einen erneuten Start ein anderes übergeordnetes Verzeichnis.
(
mkdir scala-live-demo &&
cd scala-live-demo &&
curl -fsSLo scala-cli.gz https://github.com/VirtusLab/scala-cli/releases/download/v1.17.1/scala-cli-x86_64-pc-linux.gz &&
gzip -d scala-cli.gz &&
chmod +x scala-cli &&
curl -fsSLo mediamtx.tar.gz https://github.com/bluenviron/mediamtx/releases/download/v1.21.1/mediamtx_v1.21.1_linux_amd64.tar.gz &&
tar -xzf mediamtx.tar.gz mediamtx &&
font_file=$(fc-match -f '%{file}' 'Liberation Sans') &&
test -r "$font_file" &&
cp "$font_file" font.ttf &&
./scala-cli version && ./mediamtx --version
)
Bleiben Sie für die übrigen Befehle im übergeordneten Verzeichnis. Jeder Block in runden Klammern
hält seine Verzeichniswechsel und Shell-Optionen lokal. Ein fehlgeschlagener Download kann ein
unvollständiges neues Projekt hinterlassen. Prüfen Sie es und verwenden Sie vor einem erneuten
Versuch ein neues übergeordnetes Verzeichnis. Speichern Sie das vollständige Programm unten als
scala-live-demo/Live.scala. Der erste Build lädt den Scala-Compiler und die Bibliotheken
herunter; weitere Anwendungsabhängigkeiten gibt es nicht. Die Befehle zielen explizit auf
Live.scala; ein umgebendes sbt-Projekt dient nicht als Eingabe für den Build.
FFmpeg mit Scala integrieren
Das Programm verwendet Javas ProcessBuilder mit separaten Argumenten, statt einen
Shell-Befehl durch Interpolation zusammenzusetzen. Es schreibt eine private MediaMTX-Konfiguration,
bindet HTTP an die Loopback-Schnittstelle und wählt standardmäßig verfügbare Ports. Nur die
Playlist und fertig geschriebene Segmente werden unter ihren Dateinamen ausgeliefert.
Konfiguration, Schriften und Logs sind nicht über HTTP erreichbar.
//> using scala 2.13.18
import com.sun.net.httpserver.{HttpExchange, HttpServer}
import java.net.{InetAddress, InetSocketAddress, ServerSocket}
import java.nio.file.{Files, LinkOption, Path, Paths}
import java.util.concurrent.TimeUnit.SECONDS
import scala.collection.mutable.ArrayBuffer
import scala.util.control.NonFatal
object Live {
private val children = ArrayBuffer.empty[(String, java.lang.Process)]
private var http: Option[HttpServer] = None
private var stopping = false
@volatile private var encoded = false
private def stopOwned(): Unit = synchronized {
stopping = true
http.foreach(_.stop(0))
http = None
for ((name, child) <- children.reverse if child.isAlive) {
child.destroy()
if (!child.waitFor(2, SECONDS)) {
child.destroyForcibly()
require(child.waitFor(2, SECONDS), s"Could not stop $name")
}
}
children.clear()
}
private def start(name: String, args: Seq[String], out: Path): java.lang.Process = synchronized {
require(!stopping, "Run is stopping")
val builder = new ProcessBuilder(args: _*)
.directory(out.toFile)
.redirectErrorStream(true)
.redirectOutput(out.resolve(s"$name.log").toFile)
// Ambient MediaMTX overrides must not reopen listeners disabled by our configuration.
builder.environment().keySet().removeIf(key => key.startsWith("MTX_"))
val child = builder.start()
children += name -> child
println(s"Started $name (PID ${child.pid()})")
child
}
private def serve(exchange: HttpExchange, out: Path): Unit = {
try {
val name = exchange.getRequestURI.getPath.stripPrefix("/")
val allowed = name == "stream.m3u8" || name.matches("segment-[0-9]{3}\\.ts")
val file = out.resolve(name)
if (exchange.getRequestMethod != "GET") exchange.sendResponseHeaders(405, -1)
else if (!allowed || !Files.isRegularFile(file)) exchange.sendResponseHeaders(404, -1)
else {
val contentType = if (name.endsWith(".ts")) "video/mp2t" else "application/vnd.apple.mpegurl"
exchange.getResponseHeaders.set("Content-Type", contentType)
exchange.getResponseHeaders.set("Cache-Control", "no-store")
val bytes = Files.readAllBytes(file)
exchange.sendResponseHeaders(200, bytes.length)
exchange.getResponseBody.write(bytes)
}
} finally exchange.close()
}
private def run(args: Array[String]): Unit = {
require(args.length >= 2 && args.length <= 4,
"Usage: Live <new-directory> <deadline-seconds> [rtsp-port] [http-port]")
val seconds = args(1).toInt
require(seconds >= 1 && seconds <= 120, "Deadline must be between 1 and 120 seconds")
val ports = args.drop(2).map(_.toInt)
require(ports.forall(port => port >= 0 && port <= 65535), "Invalid port")
val project = Paths.get(".").toRealPath()
val out = project.resolve(args(0)).normalize()
require(out.getParent == project, "Use a new direct child directory of this project")
require(!Files.exists(out, LinkOption.NOFOLLOW_LINKS), "Output directory already exists")
require(Files.isReadable(project.resolve("font.ttf")), "Missing readable font.ttf")
Files.createDirectory(out)
Files.copy(project.resolve("font.ttf"), out.resolve("font.ttf"))
val deadline = System.nanoTime() + SECONDS.toNanos(seconds)
def checkDeadline(): Unit =
require(System.nanoTime() < deadline, "Run deadline exceeded; output may be partial")
def alive(name: String, child: java.lang.Process): Unit =
require(child.isAlive, s"$name stopped (exit ${child.exitValue()}); inspect $name.log")
def waitUntil(ready: => Boolean)(check: => Unit): Unit = {
while (!ready) { checkDeadline(); check; Thread.sleep(100) }
checkDeadline()
}
val hook = new Thread(() => {
Console.err.println(if (encoded) "Stopped; completed HLS retained" else "Cancelled; output may be partial")
stopOwned()
})
Runtime.getRuntime.addShutdownHook(hook)
try {
println(s"Controller PID ${ProcessHandle.current().pid()}")
val requestedRtsp = ports.headOption.getOrElse(0)
val reservation = new ServerSocket(requestedRtsp, 1, InetAddress.getByName("127.0.0.1"))
val rtspPort = reservation.getLocalPort
reservation.close()
val server = synchronized {
require(!stopping, "Run is stopping")
val listener = HttpServer.create(new InetSocketAddress("127.0.0.1", ports.lift(1).getOrElse(0)), 0)
http = Some(listener)
listener.createContext("/", exchange => serve(exchange, out))
listener.start()
listener
}
val url = s"rtsp://127.0.0.1:$rtspPort/live"
val config = out.resolve("mediamtx.yml")
Files.writeString(config, s"""logLevel: info
rtsp: yes
rtspAddress: 127.0.0.1:$rtspPort
rtspTransports: [tcp]
rtmp: no
hls: no
webrtc: no
srt: no
moq: no
api: no
metrics: no
pprof: no
playback: no
paths:
live:
source: publisher
""")
val mtx = start("mediamtx", Seq(project.resolve("mediamtx").toString, config.toString), out)
def serverLog: String = Files.readString(out.resolve("mediamtx.log"))
waitUntil(serverLog.contains("[RTSP] started with listeners")) { alive("mediamtx", mtx) }
val common = Seq("ffmpeg", "-hide_banner", "-loglevel", "warning", "-nostdin")
val source = start("source", common ++ Seq(
"-re", "-f", "lavfi", "-i", "testsrc2=size=320x180:rate=10", "-t", "45",
"-an", "-c:v", "libx264", "-threads", "2", "-preset", "ultrafast",
"-tune", "zerolatency", "-g", "10", "-pix_fmt", "yuv420p",
"-f", "rtsp", "-rtsp_transport", "tcp", url), out)
waitUntil(serverLog.contains("is publishing to path 'live'")) {
alive("mediamtx", mtx); alive("source", source)
}
val filter = "scale=160:90,hue=s=0,drawbox=x=0:y=0:w=iw:h=22:color=black:t=fill," +
"drawtext=fontfile=font.ttf:text='Scala %{pts\\:hms}':fontcolor=white:fontsize=12:x=5:y=4"
val encoder = start("encoder", common ++ Seq(
"-n", "-rtsp_transport", "tcp", "-timeout", "3000000", "-i", url,
"-map", "0:v:0", "-an", "-vf", filter, "-filter_threads", "2", "-frames:v", "120",
"-c:v", "libx264", "-threads", "2", "-preset", "ultrafast", "-tune", "zerolatency",
"-pix_fmt", "yuv420p", "-g", "20", "-sc_threshold", "0", "-flags", "+cgop",
"-f", "hls", "-hls_time", "2", "-hls_playlist_type", "event",
"-hls_flags", "independent_segments+temp_file", "-hls_segment_filename", "segment-%03d.ts",
"-progress", "progress.txt", "stream.m3u8"), out)
val playlist = out.resolve("stream.m3u8")
waitUntil(Files.exists(playlist)) {
alive("mediamtx", mtx); alive("source", source); alive("encoder", encoder)
}
val playback = s"http://127.0.0.1:${server.getAddress.getPort}/stream.m3u8"
Files.writeString(out.resolve("playback.url"), playback + "\n")
println(s"Play: $playback")
waitUntil(!encoder.isAlive) { alive("mediamtx", mtx); alive("source", source) }
require(encoder.exitValue() == 0, "Encoder failed; inspect encoder.log")
val frames = Files.readString(out.resolve("progress.txt")).linesIterator
.filter(_.startsWith("frame=")).toVector.lastOption
require(frames.exists(_.stripPrefix("frame=").trim == "120"), "Incomplete video; output may be partial")
require(Files.readString(playlist).contains("#EXT-X-ENDLIST"), "Playlist was not finalized")
encoded = true
Files.writeString(out.resolve("complete.txt"), "120 frames encoded\n")
println("Encoding complete; HTTP remains available until the run deadline")
while (System.nanoTime() < deadline) Thread.sleep(100)
} finally {
stopOwned()
Runtime.getRuntime.removeShutdownHook(hook)
}
println("Completed; services stopped and HLS retained")
}
def main(args: Array[String]): Unit = {
try run(args)
catch { case NonFatal(error) => Console.err.println(s"Live demo failed: ${error.getMessage}"); sys.exit(1) }
}
}
Videostreams in Echtzeit verarbeiten
Die FFmpeg-Option -re taktet die generierte Quelle, statt sie so schnell
zu veröffentlichen, wie die CPU sie codieren kann. MediaMTX nimmt den Publisher auf
/live an; der Encoder liest denselben
RTSP-Pfad per
TCP-Transport. Dieses Beispiel verwendet RTSP, statt vorauszusetzen,
dass RTMP oder eine beliebige Kamera dasselbe Eingabeverhalten zeigt.
Der Filter ändert die Größe auf 160 mal 90 Pixel, entfernt die Farbe und zeichnet einen schwarzen
Balken mit einem Zeitstempel für die verstrichene Zeit.
drawtext benötigt Schriftunterstützung in Ihrem
FFmpeg-Build, obwohl die Schriftdatei explizit angegeben ist. Builds mit fontconfig können auf eine
andere Schrift zurückgreifen, wenn diese Datei ungültig ist. Verwenden Sie daher die bei der
Einrichtung ausgewählte installierte Schrift und prüfen Sie die sichtbare Einblendung. Ein
Videofilter erfordert Decodieren und erneutes Encoding; -c:v copy kann ihn
nicht anwenden.
Für HLS setzt der Encoder alle 20 Frames bei 10 fps einen Keyframe,
passend zur angestrebten Segmentdauer von zwei Sekunden. event behält alle
Segmente bei, und temp_file stellt jedes Segment nach Abschluss des Schreibens
unter seinem endgültigen Namen bereit. Der Durchlauf erzeugt 120 Frames und schließt anschließend
die Playlist ab. Dies ist eine begrenzte Aufnahme einer Live-Quelle, kein unbegrenztes Archiv mit
gleitendem Zeitfenster.
Erstellen Sie eine ausführbare JAR-Datei und starten Sie die Demo. --jvm system
verwendet Ihr installiertes JDK; --server=false vermeidet einen
Kompilierungsserver im Hintergrund. Die
Paketierung mit --assembly in Scala CLI schließt
die Scala-Laufzeitumgebung ein. -f ersetzt nur die Datei
live.jar dieses Beispiels, und && verhindert, dass
eine alte JAR-Datei ausgeführt wird, wenn der erneute Build fehlschlägt.
(
cd scala-live-demo &&
./scala-cli --power package Live.scala --assembly --preamble=false --server=false --jvm system -f -o live.jar &&
java -jar live.jar hls 60
)
Das Zeitlimit von 60 Sekunden beginnt nach der Argumentvalidierung und der Bereitstellung der
Schrift. Es umfasst den Dienststart, das Encoding und das Wiedergabefenster; die Kompilierung und
die Wartezeiten beim Beenden liegen außerhalb. Beide Ports werden automatisch ausgewählt.
Optionale dritte und vierte Programmargumente legen den RTSP- und den HTTP-Port fest; null fordert
einen verfügbaren Port an. Ist ein angeforderter Port belegt, schlägt der Start fehl. Jeder
Durchlauf benötigt ein neues Ausgabeverzeichnis direkt im Projekt. Ein erneuter Durchlauf mit
hls lässt daher die vorhandene Ausgabe unverändert.
FFmpeg-Integrationen testen
Wenn das Programm die Wiedergabe-URL ausgibt, öffnen Sie ein zweites Terminal im selben
übergeordneten Verzeichnis und führen Sie diesen Block aus. Der HTTP-Server ist während des
Encodings und bis zum Zeitlimit verfügbar. Sie sollten ein bewegtes Graustufenmuster mit einem
weißen Zeitstempel im oberen Balken sehen. -autoexit schließt FFplay, wenn der
abgeschlossene Stream endet. Drücken Sie q, um es früher zu schließen.
(
playback_url=$(cat scala-live-demo/hls/playback.url) &&
ffplay -autoexit "$playback_url"
)
Nachdem die Demo erfolgreich beendet wurde, enthält scala-live-demo/hls die Datei
stream.m3u8, sechs Dateien von segment-000.ts bis
segment-005.ts sowie complete.txt. Untersuchen Sie den Stream und
decodieren Sie die vollständige Aufnahme anhand ihrer tatsächlichen Playlist:
(
set -euo pipefail
cd scala-live-demo/hls &&
ffprobe -v error -show_entries stream=codec_name,width,height:format=duration \
-of json stream.m3u8 &&
ffmpeg -hide_banner -loglevel error -nostdin -xerror -progress pipe:1 -i stream.m3u8 \
-map 0:v:0 -f null - |
awk -F= '$1 == "frame" {frames=$2} END {if (frames != 120) {print "Expected 120 decoded frames" > "/dev/stderr"; exit 1}}'
)
Zu erwarten sind H.264-Video mit 160 mal 90 Pixeln und eine Playlist-Dauer von 12 Sekunden.
Analyse und Decodieren prüfen Struktur und Lesbarkeit der Aufnahme; die Wiedergabe prüft das
sichtbare Filterergebnis. Die Prüfung der Frame-Anzahl ist wichtig, weil FFmpeg ein fehlendes
HLS-Segment überspringen und trotzdem null zurückgeben kann, selbst mit
-xerror. Ein Encoder-Status von null belegt weder einen intakten Kamerafeed
noch die wahrgenommene Qualität oder die Ende-zu-Ende-Latenz.
Fehlerbehandlung und Ressourcenverwaltung
Die Steuerung prüft, ob Publisher und RTSP-Server weiterlaufen, während der Encoder arbeitet.
Ein Verbindungsabbruch kann dazu führen, dass FFmpeg mit weniger Frames endet, selbst mit
Exit-Status null. Die Fortschrittsprüfung markiert diese Aufnahme dann nicht als vollständig.
Ein Encoder-Fehler, eine gestoppte Quelle oder ein Zeitlimit, das vor Abschluss des Encodings
erreicht wird, führen zu Status eins. Logs kennzeichnen jeden Kindprozess als
mediamtx.log, source.log oder encoder.log.
Fehlgeschlagene Encoding-Durchläufe können unvollständige Medien zur Untersuchung behalten und
haben keine Datei complete.txt.
Drücken Sie Ctrl+C im ersten Terminal der Demo, um sie vorzeitig zu beenden. Der JVM-Shutdown-Hook blockiert neue Kindprozessstarts, stoppt HTTP und beendet seine Kindprozesse. Dabei wartet er bis zu zwei Sekunden pro Kindprozess, erzwingt dann dessen Beendigung und wartet weitere zwei Sekunden. Das gilt für gewöhnliche Unterbrechungs- und Beendigungssignale, nicht für ein nicht abfangbares Kill-Signal oder das Herunterfahren des Rechners. Bereits geschriebene Ausgaben bleiben erhalten; ein Abbruch während des Encodings meldet keinen Abschluss. Nach erfolgreichem Encoding bleibt die fertige Aufnahme auch bei einem vorzeitigen Stopp erhalten.
Beim regulären Zeitlimit stoppt das Programm seine Dienste und gibt Status null zurück, sofern das Encoding abgeschlossen wurde. Behalten Sie das Ausgabeverzeichnis, um die lokale Playlist erneut abzuspielen, oder entfernen Sie nach der Prüfung genau dieses Verzeichnis. Das Programm löscht oder überschreibt keinen früheren Durchlauf. Die heruntergeladenen Werkzeuge und die JAR-Datei des Projekts bleiben für einen weiteren Durchlauf mit neuem Verzeichnisnamen verfügbar.
Häufige Probleme beheben
- Wenn der Start fehlschlägt, lesen Sie das genannte Log und prüfen Sie die Voraussetzungen für ausführbare Dateien und Schriften. Das Programm stoppt bereits gestartete Kindprozesse, bevor es einen Fehler zurückgibt. Es gibt erst eine Wiedergabe-URL aus, wenn eine Playlist verfügbar ist.
- Wenn FFplay keine Verbindung herstellen kann, starten Sie es, während die Demo noch läuft.
Verwenden Sie nach deren Ende
ffplay -autoexit scala-live-demo/hls/stream.m3u8, um fertige lokale Dateien erneut abzuspielen. - Wenn das Encoding das Zeitlimit überschreitet, prüfen Sie
encoder.logund wählen Sie ein Zeitlimit von bis zu 120 Sekunden mit einem neuen Ausgabeverzeichnis. Ein längeres Zeitlimit stellt die Verbindung zu einem getrennten Publisher nicht wieder her.
Für einen anderen Filter ändern Sie den einzelnen Wert von filter,
erstellen die JAR-Datei neu und starten sie mit einem neuen Namen für das Ausgabeverzeichnis.
Behalten Sie die Thread-Limits von FFmpeg bei, während Sie messen, ob das Encoding mit der Quelle
Schritt halten kann. Zusätzliche Scala-Objekte vom Typ Future würden mehr
Encoder starten. Sie würden weder diesen Encoder beschleunigen noch diese Prozesse abbrechbar
machen. Für bereits vorhandene lokale Aufnahmen erklärt die
Scala-Anleitung zum Zusammenfügen, wie Sie kompatible Dateien ohne
erneutes Encoding zusammenfügen.