nodejs解析并下载HLS视频

2019-12-28

再上篇文章中我已经实现了用node通过解析HLS流视频中的m3u8文件, 生成了一个sh的脚本文件, 通过终端中的wget命令下载视频的分段, 然后再通过ffmpeg命令来合并hls视频为mp4的视频.

虽然功能已经实现, 也能完美下载, 但是就是这种很多小文件, 用wget这个单线程下载真的很慢, 他需要一个一个的发起请求,下载完成后再下下载下一个文件. 下载这种小文件还是并发一起下载好.

整理思路

上次我们已经把每个文件的地址都已经解析出来并且生成了一个数组, 这样来看只需要再实现一下依次请求文件地址, 把数据用fs的writeFile方法保存起来就ok.

结合到我们的 async / await 异步处理函数, 我们很简单的就能把依次下载的代码写出来

async / await 实现下载功能

直接接着上次的解析到list那开始

;(async ()=>{
  ...
  // 先来封装一个保存文件的地址
  const saveFile =(name, fileData)=>{
    return new Promise((resolve, reject)=>{
      // 这里这个fileName是m3u8文件的fileName, 文件总不能随便放嘛
      fs.writeFile(path.join(__dirname, `${fileName}/${name}`), fileData, {}, (error, data)=>{
        if(error) return reject(error)
        else resolve(data)
      })
    })
  }
  const data = await $http.get(url)
  const list = data.split('\n').filter(item=> !/^#\w+/.test(item))
  for(item of list){
    const fileData = await $http.get(filePath + item)
    await saveFile(item, fileData)
  }
})()

这样下来我倒是觉得没撒问题, 以前我去抓取文本, 存json的时候完全没问题的, 但是当我下载完打开文件的时候, 却提示我资源损坏. 这就比较难受了, 后来去搜了下, 应该是我获取的数据是文本型, 但是视频文件这种需要的是二进制的类型(渣渣一脸懵逼完全不懂), 反正按照百度上的内容改了下, 使用流来储存文件.

获取二进制数据流我们还需要对axios加一个参数, 还有下载文件我们也需要把axios的超时设置长一点

const axios = require('axios')
const fileHttp = axios.create({
  timeout: 6e5, // 这里我设置超时为10分钟
  headers,
  responseType: "stream" // 表示我们接受的数据为二进制流
})

然后我们在重新保存下文件, 当然我们也可以直接把参数给$http, 但是我们这参数比较多而且后面用的也比较多, 直接重新封装一个axios对象的话后面使用会好点嘛

...
const downFile = (url, name) => {
  return new Promise((resolve, reject)=>{
    const filePath = path.join(__dirname, `${fileName}/${name}`)
    const writer = fs.createWriteStream(filePath)
    fileHttp(url).then(({ data })=>{
      // 直接使用node二进制对象的pipe方法写入文件
      data.pipe(writer)
      writer.on("finish", e=>{
        writer.end()
        resolve(e)
      })
      writer.on("error", e=>{
        writer.end()
        reject(e)
      })
    }).catch(reject)
  })
}

for(item of list){
  await downFile(filePath + item, item)
}

这下子下载完成我终于能播放出视频了, 不过虽然做到了批量下载不过还是一个一个的慢慢下载的

并行下载

先来想一下我们的需求, 就是要同时下载这些文件, 但是我们却不能直接把全部下载任务都丢给异步全部一起下载(浏览器对同时发起的xhr请求有限制, nodejs中我不清楚但是几百条请求出去大半也会很多请求丢包), 需要控制同时下载的数量.

想要实现控制那么多请求, 并且前面一旦有完成再接下来的数据, 我先写一个Promise模块, then后再调用自己处理队列的内容, 队列内容处理完成之后就停止调用自己

...
// list []
...

let doneIndex = 1
const primiseArr = []

// 获取资源并保存
const getRes = (item) =>{
  return new Promise((resolve, reject)=>{
    const url = filePath + item
    downFile(filePath + item, item)
      .then(()=>resolve({url, item}))
      .catch(e=>reject({url, item, e}))
  })
}
// 任务调度核心
const resCore =()=>{
  const item = list.shift()
  if(!item) return console.log(doneIndex++, "done !")
  primiseArr.push(getRes(item).then(e=>{
    console.log(e.item, 'done')
    resCore()
  }).catch(e=>{
    console.log(e)
    resCore()
  }))
}

队列下载都已经好了. 现在就差我们把最开始的数量调用出去, 完成一个任务就会从队列中取到下个任务去完成

我们直接申请一个num, 然后while(num–), 这样我们就选择num次调用了嘛

1
2
3

// 这里我们设置一个控制的数据
let coreNum = 6
while(coreNum--) resCore()

全部代码

老规矩到最后还是把全部代码贴上来

const $http = require('./libs/http')
const fs = require('fs')
const path = require('path')
const axios = require('axios')
const fileHttp = axios.create({
  timeout: 6e5, // 这里我设置超时为10分钟
  headers,
  responseType: "stream" // 表示我们接受的数据为二进制流
})

const url = 'http://.../index.m3u8'

// 这里我使用 await 来处理异步, 所以申请了一个 async 的匿名函数
;(async ()=>{
  // 解析链接
  const { filePath, fileName } = ((url)=>{
    const arr = url.split('/')
    return {
      "fileName": arr[arr.length - 1].split('.')[0],
      "filePath": url.replace(arr[arr.length-1],'')
    }
  })(url)
  const downFile = (url, name) => {
    return new Promise((resolve, reject)=>{
      const filePath = path.join(__dirname, `${fileName}/${name}`)
      const writer = fs.createWriteStream(filePath)
      fileHttp(url).then(({ data })=>{
        // 直接使用node二进制对象的pipe方法写入文件
        data.pipe(writer)
        writer.on("finish", e=>{
          writer.end()
          resolve(e)
        })
        writer.on("error", e=>{
          writer.end()
          reject(e)
        })
      }).catch(reject)
    })
  }
  const data = await $http.get(url)
  const list = data.split('\n').filter(item=> !/^#\w+/.test(item))
  
  let doneIndex = 1
  const primiseArr = []
  
  // 获取资源并保存
  const getRes = (item) =>{
    return new Promise((resolve, reject)=>{
      const url = filePath + item
      downFile(filePath + item, item)
        .then(()=>resolve({url, item}))
        .catch(e=>reject({url, item, e}))
    })
  }
  // 任务调度核心
  const resCore =()=>{
    const item = list.shift()
    if(!item) return console.log(doneIndex++, "done !")
    primiseArr.push(getRes(item).then(e=>{
      console.log(e.item, 'done')
      resCore()
    }).catch(e=>{
      console.log(e)
      resCore()
    }))
  }
  let coreNum = 6
  while(coreNum--) resCore()
})()