使用Golang簡單實現(xiàn)七牛圖片處理API
之前一直在用qiniu的存儲服務，生成圖片的縮略圖，模糊圖，視頻的webp，現(xiàn)在需要把存儲移到s3上，那么這些圖片，視頻處理就要自己動手寫了，本文梳理一下大致的思路。
分析需求
先看一下qiniu的接口是如何處理圖片的，例如先截取視頻第一秒的圖片，再把圖片縮略，最后存儲到一個新的key，命令可以這么寫 vframe/jpg/offset/1|imageMogr2/thumbnail/400x|saveas/xxx, 可以看到三個操作之間用 | 符號分割，類似unix 的 pipe 操作。
上面的操作算作一個cmd, 一次API請求可以同時處理多個cmd，cmd之間用分號分割, 處理完畢后，在回調中把處理結果返回，例如
{
"id": "xxxxx",
"pipeline": "xxx",
"code": 0,
"desc": "The fop was completed successfully",
"reqid": "xTsAAFnxUbR5J10U",
"inputBucket": "xxx",
"inputKey": "xxxxx",
"items": [
{
"cmd": "vframe/jpg/offset/1|imageMogr2/thumbnail/400x|saveas/ZmFtZS1wcml2YXRlOm1vbWVudC9jb3Zlci9zbmFwL3ZpZGVvL2M5YzdjZjQ5LTU3NGQtNGZjMS1iZDFkLTRkYjZkMzlkZWY1Ni8wLzA=",
"code": 0,
"desc": "The fop was completed successfully",
"hash": "FhdN6V8EI4vW4XJGALSfxutvMEIv",
"key": "xx",
"returnOld": 0
},
{
"cmd": "vframe/jpg/offset/1|imageMogr2/thumbnail/400x|imageMogr2/blur/45x8|saveas/ZmFtZS1wcml2YXRlOm1vbWVudC9jb3Zlci9zbmFwL3ZpZGVvL2M5YzdjZjQ5LTU3NGQtNGZjMS1iZDFkLTRkYjZkMzlkZWY1Ni8wLzBfYmx1cg==",
"code": 0,
"desc": "The fop was completed successfully",
"hash": "FgNiRzrCsa7TZx1xVSb_4d5TiaK3",
"key": "xxx",
"returnOld": 0
}
]
}
分解需求
這個程序大致需要這么幾個部分:
一個http接口，接受任務，接受后把任務扔到隊列，返回一個job ID。 worker異步處理任務，worker的個數(shù) 和 每個worker 并行的處理的個數(shù) 能夠配置，worker有重試機制。
從 job payload 中解析出需要做的任務，解析出每個cmd, 最好能并行執(zhí)行每一個 cmd, 記錄每一個cmd的結果
每個cmd中有多個 operation, 并且用 pipe 連接，前一個operaion的輸出是后一個operation的輸入
可以把 1 和 2，3 分開來看，1 比較獨立，之前寫過一個worker的模型，參考的是這篇文章 Handling 1 Million Requests per Minute with Go，比較詳細，是用 go channel 作為queue的，我加了一個 beanstalk 作為 queue的 providor。還有一點改進是，文章中只提供了worker數(shù)量的設置，我再加了一個參數(shù)，設定每個worker可以并行執(zhí)行的協(xié)程數(shù)。所以下面主要講講3， 2的解決辦法
Pipe
可以參考這個庫 pipe, 用法如下：
p := pipe.Line(
pipe.ReadFile("test.png"),
resize(300, 300),
blur(0.5),
)
output, err := pipe.CombinedOutput(p)
if err != nil {
fmt.Printf("%v\n", err)
}
buf := bytes.NewBuffer(output)
img, _ := imaging.Decode(buf)
imaging.Save(img, "test_a.png")
還是比較方便的，建一個 Cmd struct, 利用正則匹配一下每個 Operation 的參數(shù)，放入一個 []Op slice, 最后執(zhí)行，struct和方法如下：
type Cmd struct {
cmd string
saveas string
ops []Op
err error
}
type Op interface {
getPipe() pipe.Pipe
}
type ResizeOp struct {
width, height int
}
func (c ResizeOp) getPipe() pipe.Pipe {
return resize(c.width, c.height)
}
//使用方法
cmdStr := `file/test.png|thumbnail/x300|blur/20x8`
cmd := Cmd{cmdStr, "test_b.png", nil, nil}
cmd.parse()
cmd.doOps()
sync.WaitGroup
單個cmd處理解決后，就是多個cmd的并行問題，沒啥好想的，直接用 sync.WaitGroup 就可以完美解決。一步一步來，我們先看看這個struct的使用方法：
func main() {
cmds := []string{}
for i := 0; i < 10000; i++ {
cmds = append(cmds, fmt.Sprintf("cmd-%d", i))
}
results := handleCmds(cmds)
fmt.Println(len(results)) // 10000
}
func doCmd(cmd string) string {
return fmt.Sprintf("cmd=%s", cmd)
}
func handleCmds(cmds []string) (results []string) {
fmt.Println(len(cmds)) //10000
var count uint64
group := sync.WaitGroup{}
lock := sync.Mutex{}
for _, item := range cmds {
// 計數(shù)加一
group.Add(1)
go func(cmd string) {
result := doCmd(cmd)
atomic.AddUint64(&count, 1)
lock.Lock()
results = append(results, result)
lock.Unlock()
// 計數(shù)減一
group.Done()
}(item)
}
// 阻塞
group.Wait()
fmt.Printf("count=%d \n", count) // 10000
return
}
group本質大概是一個計數(shù)器，計數(shù) > 0時, group.Wait() 會阻塞，直到 計數(shù) == 0. 這里還有一點要注意，就是 results = append(results, result) 的操作是線程不安全的，清楚這里 results 是共享的，需要加鎖來保證同步，否則最后 len(results) 不為 10000。
我們建一個BenchCmd， 來存放 cmds. 如下：
type BenchCmd struct {
cmds []Cmd
waitGroup sync.WaitGroup
errs []error
lock sync.Mutex
}
func (b *BenchCmd) doCmds() {
for _, item := range b.cmds {
b.waitGroup.Add(1)
go func(cmd Cmd) {
cmd.parse()
err := cmd.doOps()
b.lock.Lock()
b.errs = append(b.errs, err)
b.lock.Unlock()
b.waitGroup.Done()
}(item)
}
b.waitGroup.Wait()
}
最后的調用就像這樣：
var cmds []Cmd
cmd_a := Cmd{`file/test.png|thumbnail/x300|blur/20x8`, "test_a.png", nil, nil}
cmd_b := Cmd{`file/test.png|thumbnail/500x1000|blur/20x108`, "test_b.png", nil, nil}
cmd_c := Cmd{`file/test.png|thumbnail/300x300`, "test_c.png", nil, nil}
cmds = append(cmds, cmd_a)
cmds = append(cmds, cmd_b)
cmds = append(cmds, cmd_c)
bench := BenchCmd{
cmds: cmds,
waitGroup: sync.WaitGroup{},
lock: sync.Mutex{},
}
bench.doCmds()
fmt.Println(bench.errs)
這只是一個初級的實驗，思考還不夠全面，并且只是模仿API，qiniu應該不是這么做的，耦合更低，可能各個Cmd都有各自處理的集群，那pipe這個庫就暫時沒法解決了，目前的局限在于 每個Cmd必須都在一個進程中。
相關文章
本文主要介紹了go語言使用jwt認證的實現(xiàn)，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-04-04
這篇文章主要介紹了golang?http請求未釋放造成的錯誤問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-01-01
go語言中有一個非常神奇的函數(shù)init,它可以在所有程序執(zhí)行開始前被執(zhí)行,并且每個package下面可以存在多個init函數(shù),這篇文章主要給大家介紹了關于Go語言中init函數(shù)特點、用途和注意事項的相關資料,需要的朋友可以參考下2023-07-07
這篇文章主要為大家介紹了GO語言基本類型String和Slice,Map操作示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-08-08
本文主要介紹了Go語言什么情況下應該使用指針，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-01-01![go語言中[]*int和*[]int的具體使用](http://img.jbzj.com/images/xgimg/bcimg5.png)
本文主要介紹了go語言中[]*int和*[]int的具體使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-04-04
這篇文章主要介紹了Go 語言的指針的學習筆記，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-09-09
switch 語句提供了一種簡潔的方式來執(zhí)行多路分支選擇,本文主要介紹了Golang switch語句的具體使用,具有一定的參考價值,感興趣的可以了解一下2024-08-08
Base64是網絡上最常見的用于傳輸8Bit字節(jié)代碼的編碼方式之一,可用于在HTTP環(huán)境下傳遞較長的標識信息,下面這篇文章主要給大家介紹了關于Go語言中的Base64編碼原理介紹以及使用的相關資料,需要的朋友可以參考下2022-01-01
使用client-go工具調用kubernetes API接口的教程詳解(v1.17版本)
這篇文章主要介紹了使用client-go工具調kubernetes API接口(v1.17版本),本文通過圖文實例相結合給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-08-08