這篇文章主要介紹了基于Serverless技術的視頻截幀架構方法怎么實現的相關知識，內容詳細易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇基于Serverless技術的視頻截幀架構方法怎么實現文章都會有所收獲，下面我們一起來看看吧。

創新互聯建站-專業網站定制、快速模板網站建設、高性價比恩陽網站開發、企業建站全套包干低至880元,成熟完善的模板庫,直接使用。一站式恩陽網站制作公司更省心,省錢,快速模板網站建設找我們，業務覆蓋恩陽地區。費用合理售后完善，十年實體公司更值得信賴。

視頻截幀需求概述

直播行業受到越來越多的法律、法規和政策的規限，在行業一般標準和運營規程的約束下，每一個直播平臺都有義務對非法的直播內容，以及主播與觀眾之間的不當互動采取措施，為直播行業更為規范的發展做出貢獻。如何第一時間監控到直播流中的非法內容，是直播平臺需要面對的共同挑戰，視頻截幀就是滿足內容審核需求的常規操作。視頻截幀可以根據視頻直播的不同風險等級，選擇不同的頻率對直播流進行截幀處理，保存后的圖片可以統一上傳到自建或第三方內容審核平臺，用于涉黃、涉政、廣告等場景的識別。除此之外，某些特定的業務需求也需要通過視頻截幀來實現，比如在線課堂類應用對學生的聽課狀態進行智能分析等。

視頻截幀技術架構分析

對于視頻流的截幀操作，可以通過FFmpeg命令實現。FFmpeg的截幀命令使用非常簡單，每次截取一張圖片后，可以將圖片上傳到對象存儲OSS，同時將對應的截幀信息發送到消息隊列Kafka。這樣審核服務（可以是第三方服務或是自建服務）就可以從Kafka獲取截幀信息，并從OSS拉取對應的圖片進行處理。在這個架構中，引入Kafka是為了通過異步處理機制緩解審核服務在業務高峰期的負載。

FFmpeg使用雖然簡單，但是這是一個對于CPU計算力需求量非常大的操作。如果按照1秒的固定頻率對視頻流進行截幀操作，1臺16核的ECS大概能同時承擔100路視頻流的截幀任務。為了確保業務高峰期的服務穩定，就需要準備大量ECS來部署視頻截幀服務。而大多數互聯網應用都存在明顯的波峰波谷，比如每天晚上的黃金時間是業務高峰，而24點以后的業務量會呈明顯下降的趨勢。這樣的業務波動對整體的資源規劃帶來了極大的挑戰，如果按照固定的ECS集群規模來部署截幀服務，會存在兩個非常明顯的弊端：

1. 為了支持業務高峰，必須按照高峰期的用戶量來評估集群規模，在業務低峰期就會造成巨大的浪費。

2. 在某些場景下，比如明星效應的帶動，業務量會有突增，有可能需要對集群進行臨時擴容，這種情況下往往擴容速度會滯后于業務流的增速，造成部分業務的降級處理。

為了更好地提升資源利用率，也可以通過彈性ECS實例配合容器化的方式部署應用，以實現集群規模動態適配真實業務量的變化。但在實際情況中，這樣的方案彈性伸縮策略實現比較復雜，彈性伸縮能力相對滯后，效果可能并不會太好。其中的根本原因是在傳統的服務架構中，一個應用啟動后都是長期保持運行，在運行期間會并發處理多個業務需求，不管業務量如何變化，這個應用占據的計算力都不會有本質的變化。

有沒有一種直截了當的方式，可以在一路直播視頻流開啟后，拉起對應的計算力承接截幀任務，而在視頻流關閉后，自動將計算力釋放呢？這樣的方式不需要應用實例長駐，可以實現真正的計算資源按需分配，也不需要借助額外的手段動態調整截幀服務的集群規模，是一種最為理想的方案。

作為云原生Serverless技術的代表，阿里云函數計算FC就正好實現了這樣的思路。

基于函數計算FC的Serverless架構

函數計算FC是事件驅動的全托管計算服務。使用函數計算，用戶無需采購與管理服務器等基礎設施，只需編寫并上傳代碼。函數計算會自動準備好計算資源，彈性地、可靠地運行任務，并提供日志查詢、性能監控和報警等功能。借助函數計算FC，可以快速構建任何類型的應用和服務，并且只需為任務實際消耗的資源付費。

函數計算FC提供了一種事件驅動的計算模型，函數的執行是由事件驅動。函數的執行可以通過函數使用者自己觸發，也可以由其它一些事件源來觸發。可以在指定函數中創建觸發器，該觸發器描述了一組規則，當某個事件滿足這些規則，事件源就會觸發相應的函數。比如對于HTTP觸發而言，用戶的一次HTTP請求就能觸發一個函數；而對于OSS觸發器而言，OSS上新增或修改一個文件就能觸發一個函數。在視頻截幀場景中，函數只需要在每一個直播流開始推送之前，通過業務程序主動觸發一個截幀函數就可以了。因此原有的截幀架構只需要做很小的調整，就能遷移到函數計算平臺上來，以享受Serverless的價值。

Serverless架構視頻截幀技術實現

現在，我們通過幾個簡單的步驟，來搭建基于函數計算FC的Serverless架構，以實現視頻截幀需求。函數計算FC對于Node.js、Python、PHP、Java等多種語言提供了原生的運行環境，特別是像Python這樣的腳本語言，可以實現在函數計算平臺上直接修改調度代碼，使用非常簡單，因此本文的示例代碼通過Python來實現。

當然，函數計算FC對于開發語言沒有要求，任何主流的開發語言都可以很好的支持。通過函數計算FC提供的Custom Runtime，可以為任務語言建立自定義的運行環境。Custom Runtime本質上是一個HTTP Server，這個HTTP Server接管了函數計算系統的所有請求，包括來自事件調用或者HTTP函數調用。

輸出視頻流

我們完全可以通過第三方的視頻流服務進行開發，但為了更方便地在本地進行調試，可以通過自建RTMP服務實現視頻流的輸出。其中比較簡單的方式是購買1臺ECS，并部署Nginx實現RTMP服務，這需要加載nginx-rtmp-module模塊，我們可以在互聯網上找到很多相關的教程，本文不再贅述。

有了RTMP服務之后，我們就可以去http://ffmpeg.org/下載編譯好的ffmpeg程序包，通過FFmpeg命令讓本地的視頻文件推送到RTMP服務。比如用如下的方式：

ffmpeg -re -i test.flv -vcodec copy -acodec aac -ar 44100 -f flv rtmp://xxx.xxx.xxx.xxx:1935/stream/test

接下來，我們打開瀏覽器，輸入對應的RTMP直播地址，就能拉起對應的播放器觀看直播了。rtmp://xxx.xxx.xxx.xxx:1935/stream/test

安裝Funcraft

Funcraft是一個支持Serverless應用部署的工具，可以幫助用戶便捷地管理函數計算、API網關、日志服務等資源。Funcraft通過一個資源配置文件template.yml，就能實現開發、構建、部署等操作，能夠在我們使用函數計算FC實現Serverless架構的過程中，極大程度的減少配置和部署工作量。

有三種方式可以安裝Funcraft，包括npm包管理安裝、下載二進制安裝，以及Homebrew包管理器安裝。對于沒有安裝npm的環境而言，最簡單的方式是通過下載二進制安裝。我們可以通過https://github.com/alibaba/funcraft/releases下載對應平臺的Funcraft安裝包，解壓后就可以使用。可以通過以下命令檢驗Funcraft包是否安裝成功：

fun --version

如果執行命令后返回Funcraft對應的版本號，比如3.6.20，那就代表安裝成功了。

在第一次使用fun之前需要先執行 fun config 命令進行初始化配置，這個操作需要提供阿里云 Account ID、Access Key Id、Secret Access Key、 Default Region Name等常規信息， 這些信息可以從函數計算控制臺首頁的右上方獲得。其他的信息比如timeout等直接使用默認值即可。

配置OSS

由于截幀后保存的文件要上傳到對象存儲OSS備用，我們需要開通阿里云OSS服務，并創建對應的Bucket，具體的操作我們可以參考https://www.aliyun.com/product/oss完成。

配置日志服務SLS

日志服務SLS（Log Service）是阿里云提供的針對日志類數據的一站式服務，通過日志服務存儲函數日志需要在函數對應的服務中配置日志項目和日志倉庫，并授予該服務訪問日志服務的權限。函數日志會打印到配置的日志倉庫中，同一個服務下的所有函數日志都會打印到同一個日志倉庫中?？梢詫⒑瘮祱绦械娜罩敬鎯χ涟⒗镌迫罩痉?，再根據日志服務中存儲的函數日志來執行代碼調試、故障分析、數據分析等操作。

我們可以參考創建日志項目和日志倉庫來配置日志服務SLS，要確保日志項目和日志倉庫都已經成功創建，在部署函數的時候，需要使用到日志項目和日志倉庫的信息。

編寫函數

現在我們通過一段最簡單的Python代碼，來體驗如何通過函數計算FC實現截幀操作，為了讓讀者理解起來更輕松，我們暫時將業務邏輯簡化，只做如下兩個動作：

1. 通過FFmpeg命令截取1張圖片

2. 保存到OSS

import json, oss2, subprocess

HELLO_WORLD = b'Snapshot OK!\n'
OSS_BUCKET_NAME = b'snapshot'

def handler(environ, start_response):
    logger = logging.getLogger() 
    context = environ['fc.context']
    request_uri = environ['fc.request_uri']
    for k, v in environ.items():
        if k.startswith('HTTP_'):
            pass
    try:        
        request_body_size = int(environ.get('CONTENT_LENGTH', 0))
    except (ValueError):        
        request_body_size = 0
    #獲得直播流的地址
    rtmp_url = request_body.decode("UTF-8")
    #通過FFmpeg命令截取一張圖片
    cmd = ['/code/ffmpeg', '-i', rtmp_url, '-frames:v', '1', '/tmp/snapshot.png' ]
    try:
        subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
    except subprocess.CalledProcessError as exc:
        err_ret = {'returncode': exc.returncode, 'cmd': exc.cmd, 'output': exc.output.decode(),'stderr': exc.stderr.decode()}
        print(json.dumps(err_ret))
    raise Exception(context.request_id + ' transcode failure')
    #上傳到OSS
    creds = context.credentials
    auth = oss2.StsAuth(creds.access_key_id, creds.access_key_secret, creds.security_token)
    bucket = oss2.Bucket(auth, 'http://oss-{}-internal.aliyuncs.com'.format(context.region), OSS_BUCKET_NAME)
    logger.info('upload pictures to OSS ...')
    for filename in os.listdir("/tmp"): 
        bucket.put_object_from_file("example/" + filename, "/tmp/" + filename)
    status = '200 OK'
    response_headers = [('Content-type', 'text/plain')]
    start_response(status, response_headers)
    return [HELLO_WORLD]

讓我們分析一下這段代碼。首先，除了Python的標準模塊，函數計算FC的Python運行環境中還包含了一些常用模塊，其實就包括了oss2，用于在函數中操作阿里云對象存儲OSS。因此，我們可以直接在代碼中引入oss2這個模塊。

函數計算FC集成了多種類型的觸發器，這個示例函數使用的是HTTP觸發器，每一個HTTP請求都會觸發一個函數的執行。對于使用HTTP觸發器的Python代碼，入口函數就是handler，其中的environ參數攜帶了調用函數的客戶端相關信息以及上下文信息。我們可以從HTTP請求Body中，解析出STMP直播流的地址，并通過FFmpeg命令截取一張圖片。

在這段代碼中，ffmpeg可執行程序位于/code目錄，可以通過/code/ffmpeg路徑進行執行。這是因為我們在對函數進行部署的時候，已經將ffmpeg可執行程序和這段代碼打包在了這個目錄中，在接下來介紹函數部署的時候，我們會進一步介紹如何將函數代碼與可執行程序一起打包。

在對/tmp目錄保存的圖片文件上傳到OSS的過程中，我們可以直接從函數上下文中獲取訪問OSS的憑證，這樣就不需要再通過配置文件拿到accessKey，accessSecret等信息，從而減少工作量。

部署函數

首先，我們在本地創建一個工作目錄，并在這個目錄下創建一個名為code的子目錄，將Linux環境的ffmpeg可執行文件復制到code目錄中，這樣可以在代碼中通過路徑/code/ffmpeg調用ffmpeg命令。

接下來，開始最重要的工作，在當前工作目錄中創建template.yml文件，描述所有的部署信息。

ROSTemplateFormatVersion: '2015-09-01'
Transform: 'Aliyun::Serverless-2018-04-03'
Resources:
  #服務
  snapshotService:
    Type: 'Aliyun::Serverless::Service'
    Properties:
      Description: 'Snapshot Semo'
      Policies:
        - AliyunOSSFullAccess
      #之前創建的日志項目和日志倉庫
      LogConfig:
        Project: fc-bj-pro
        Logstore: fc-log
    #函數
    snapshot:
      Type: 'Aliyun::Serverless::Function'
      Properties:
        Handler: index.handler
        Runtime: python3
        MemorySize: 128
        Timeout: 600
        CodeUri: './code'
      # HTTP觸發器
      Events:
        http-test:
          Type: HTTP
          Properties:
            AuthType: ANONYMOUS
            Methods: ['POST']

配置信息比較簡單，我們需要先定義一個服務。服務是函數計算資源管理的單位。從業務場景出發，一個應用可以拆分為多個服務。從資源使用維度出發，一個服務可以由多個函數組成。例如一個數據處理服務，分為數據準備和數據處理兩部分。數據準備函數資源需求小，可以選擇小規格實例。數據處理函數資源需求大，可以選擇大規格實例。創建函數前必須先創建服務，同一個服務下的所有函數共享一些相同的設置，例如服務授權、日志配置。在這段代碼中，我們創建的服務名為snapshotService，其擁有對OSS的全部操作權限，并引用了之前所創建的日志項目和日志倉庫。

在函數實例規格的配置上，由于每個計算實例只需要處理一路視頻流，我們選擇最低的規格，也就是128M內存的實例即可。

接下來，我們要定義一個函數，配置其對應的運行環境、入口方法、代碼目錄、超時時間等信息，并為這個函數定義一個HTTP觸發器。在這段代碼中，函數名為snapshot，對應的運行環境為Python3，并且定義了一個名為http-test的HTTP觸發器。

在這個工作目錄，執行fun deploy，如果看到提示server SnapshotService deploy success，就代表代碼和ffmpeg程序已經打包部署到云上了。

在控制臺的服務與函數菜單，我們可以看到上傳的服務以及函數信息，甚至可以在線查看和修改函數代碼。

執行函數

由于這是一個HTTP類型的函數，我們可以通過curl命令或其他HTTP工具比如Postman向函數計算FC發起一次HTTP請求，驗證截幀操作的執行結果。當然，函數計算FC控制臺也提供了一個可視化操作界面來對函數進行驗證，在這個界面可以快速發起一次HTTP請求。

如果函數執行成功，我們就可以前往OSS控制臺檢查截取好的圖片是否已經成功上傳。至此，我們已經搭建好最基本的Serverless視頻截幀架構，可以通過HTTP請求觸發函數計算對視頻流截取一張圖片，并上傳到OSS。

連續截幀

單張圖片的截幀操作非常簡單，在FFmpeg命令執行完成后，就可以直接將臨時文件夾中的圖片上傳到OSS，然后完成函數的生命周期。單張圖片截幀已經可以滿足很多種業務場景，但如果需要按照固定頻率進行連續截幀，并實時將保存好的圖片上傳到OSS，就需要對代碼做一些修改。

配置消息隊列Kafka

為了降低內容審核服務在業務高峰期的工作負荷，我們可以在截幀服務和內容審核服務中間引入消息隊列Kafka，這樣內容審核服務就能通過消費從Kafka收到的消息，對保存的圖片進行異步處理。在視頻截幀架構中，Kafka起到了非常重要的信息中轉作用，直播的并發越大，截幀頻率越高，Kafka所承受的壓力就會越大。特別是在業務高峰期，需要讓Kafka在高負荷的工作中保持穩定性，直接使用阿里云提供的消息隊列Kafka能夠幫助我們大幅減少Kafka集群的維護工作量，用最簡單的方式獲得可以動態擴展的高可用Kafka服務。

我們可以打開Kafka開通界面，根據實際場景的需求購買對應規格的Kafka實例。在Kafka控制臺的基本信息中可，我們可以看到Kafka實例對應的默認接入點。

接下來，我們進入Topic管理界面，創建一個用于截幀服務的Topic。

Kafka實例的默認接入點和Topic名稱是我們需要在后續步驟中使用到的信息。

安裝Kafka客戶端SDK

在此之前，我們還需要通過一些額外的操作，獲取函數對Kafka的寫入能力。

因為需要使用到Kafka SDK，我們可以通過Funcraft工具結合Python包管理工具pip進行kafka SDK模塊的安裝：

fun install --runtime python3 --package-type pip kafka-python

執行命令后有如下提示信息：

此時我們會發現在目錄下會生成一個.fun文件夾 ，我們安裝的依賴包就在該目錄下：

打通對VPC內資源的訪問能力

默認情況下，函數計算無法訪問VPC中的資源，由于我們需要讓函數訪問部署在VPC內的Kafka服務需要手動為服務配置VPC功能和相關權限。我們可以參考配置函數訪問VPC內資源，打通函數與Kafka服務之間的連接，其原理就是通過授予

彈性網卡ENI訪問VPC的權限，并將此彈性網卡ENI插入到執行函數的實例上，從而使函數可以訪問您VPC內的資源。

代碼實現

可以通過如下FFmpeg命令實現按照指定頻繁的連續截幀：

ffmpeg -i rtmp://xxx.xxx.xxx.xxx:1935/stream/test -r 1 -strftime 1 /tmp/snapshot/%Y%m%d%H%M%S.jpg

在命令運行的過程中，Python程序當前進程會等待視頻流推送結束，因此我們需要修改函數代碼，啟動一個新的掃描進程。掃描進程不斷檢查圖片目錄，一旦發現有新的圖片生成，就將圖片上傳到OSS，同時將截幀信息發送到Kafka，最后將圖片從圖片目錄中刪除。

import logging, json, oss2, subprocess
from multiprocessing import Process
from kafka import KafkaProducer

HELLO_WORLD = b'Snapshot OK!\n'
OSS_BUCKET_NAME = b'snapshot'
logger = logging.getLogger()
output_dir = '/tmp/shapshot'
# 掃描圖片目錄
def scan(bucket, producer):
    flag = 1
    while flag:
        for filename in os.listdir(output_dir):
            if filename == 'over':
                # ffmpeg命令完成，準備停止掃描
                flag = 0
                continue
            logger.info("found image: %s", snapshotFile)
            try: 
                full_path = os.path.join(output_dir, filename)
                # 上傳到OSS
                bucket.put_object_from_file("snapshot/" + filename, full_path)
                # 發送到Kafka
                producer.send('snapshot', filename.encode('utf-8'))
                # 刪除圖片
                os.remove(full_path)
            except Exception as e:
                logger.error("got exception: %s for %s", e.message, filename)
        time.sleep(1)

def handler(environ, start_response):
    logger = logging.getLogger() 
    context = environ['fc.context']
    #創建圖片輸出文件夾
    if not os.path.exists(output_dir):
        os.mkdir(output_dir)
    #解析HTTP請求，獲得直播流的地址
    request_uri = environ['fc.request_uri']
    for k, v in environ.items():
        if k.startswith('HTTP_'):
            pass
    try:        
        request_body_size = int(environ.get('CONTENT_LENGTH', 0))
    except (ValueError):        
        request_body_size = 0
    rtmp_url = request_body.decode("UTF-8")
    #啟動Kafka Producer
    producer = KafkaProducer(bootstrap_servers='XX.XX.XX.XX:9092,XX.XX.XX.XX:9092')
    #啟動OSS Bucket
    creds = context.credentials
    auth = oss2.StsAuth(creds.access_key_id, creds.access_key_secret, creds.security_token)
    bucket = oss2.Bucket(auth, 'http://oss-{}-internal.aliyuncs.com'.format(context.region), OSS_BUCKET_NAME)
    #啟動掃描進程
    scan_process = Process(target=scan, args=(bucket, producer))
    #通過FFmpeg命令按每秒1幀的頻繁連續截幀
    cmd = ["/code/ffmpeg", "-y", "-i", rtmp_url, "-f", "image2", "-r", "1",
        "-strftime", "1", os.path.join(output_dir, "%Y%m%d%H%M%S.jpg")]
    try:
        subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
    except subprocess.CalledProcessError as exc:
        err_ret = {'returncode': exc.returncode, 'cmd': exc.cmd, 'output': exc.output.decode(),'stderr': exc.stderr.decode()}
        logger.error(json.dumps(err_ret))
    raise Exception(context.request_id + ' transcode failure')
    #寫入標志文件，子進程結束工作
    os.system("touch %s" % os.path.join(output_dir, 'over'))
    scan_process.join()
    producer.close()
    status = '200 OK'
    response_headers = [('Content-type', 'text/plain')]
    start_response(status, response_headers)
    return [HELLO_WORLD]

進一步優化

長視頻截幀

函數計算FC默認的彈性實例有600秒，也就是10分鐘的函數執行時長上限，也就是說，一個函數在觸發后，如果運行了10分鐘還沒有完成計算任務，會自動退出。這個限制會影響播放時間大于10分鐘以上的視頻流截幀操作，長視頻是非常普遍的，如何繞過這個限制對長視頻進行截幀處理呢？我們可以通過如下三種方案解決：

1. 每個函數只截1幀：當截幀頻率比較低，或者只需要在某幾個特定的時間點對視頻流進行截幀的時候，我們不需要讓函數的生命周期與視頻流的播放周期保持一致，可以讓每一個函數在啟動后，只截取單幀圖片。通過自定義的觸發程序，可以在必要的時間點啟動函數，也可以通過Serverless工作流來對函數進行更復雜的編排。

2. 通過多個函數接力完成：函數計算FC內置了fc2模塊，可以用于函數之間的相互調用。這樣我們可以控制每個截幀函數的運行時間控制在10分鐘之內，比如8分鐘為固定的運行周期。在一個函數結束前，啟動另一個函數接力完成截幀任務，直到視頻流結束。這種方案非常適合對于截幀頻率的精確度要求不是特別高的場景，因為在兩個函數進行任務交接的時候，會有一秒左右的時間無法嚴格保證截幀頻率的精確度。

3. 使用性能實例：除了默認的彈性實例以外，函數計算FC還提供了性能實例，性能實力屬于大規格實例，資源上限更高，適配場景更多，能夠突破10分鐘的執行時長上限。性能實例擴容速度較慢，彈性伸縮能力不及彈性實例，但我們可以通過單實例多并發和預留模式的配合，來提升性能實例的彈性能力。具體介紹可以參考單實例多并發和預留模式。

費用優化

函數計算提供了豐富的計量模式、有競爭力的定價，以及詳細的資源使用指標，結合Serverless以應用為中心的架構，讓資源管理前所未有的便捷，在不同場景下都能獲得極具競爭力的成本。

根據對資源的規格和彈性要求的差異，函數計算提供了預付費（包年包月）和后付費（按量付費）兩種計量模式。在常規情況下，只需要使用按量付費模式，只需為實際使用的函數計算資源付費，不需要提前購買資源。但用戶可以根據每天實際的資源使用情況，靈活選擇預付費模式節省使用成本。預付費模式是指用戶預先購買一定時長的計算力，在預購計算力的生命周期內，可以逐秒抵扣函數運行時所消耗的資源，而預付費模式的單價是永小于后付費模式的。

在函數計算控制臺的資源中心頁面，能夠一目了然地看到當前賬戶下的資源實際使用情況，包括資源使用中穩定和彈性的部分，通過這些信息，能夠合理分配的預付費和后付費資源。在資源使用詳情圖中，綠色曲線為每天的實際資源使用量，黃色直線代表其中可以被預付費資源抵扣的使用量，我們可以根據實際情況適當的提升預付資源的占比，使更多的資源使用量被預付費資源覆蓋，從而降低整理的資源費用。

關于“基于Serverless技術的視頻截幀架構方法怎么實現”這篇文章的內容就介紹到這里，感謝各位的閱讀！相信大家對“基于Serverless技術的視頻截幀架構方法怎么實現”知識都有一定的了解，大家如果還想學習更多知識，歡迎關注創新互聯行業資訊頻道。

名稱欄目：基于Serverless技術的視頻截幀架構方法怎么實現
網址分享：http://m.jcarcd.cn/article/isgepd.html