[TOC] ## 概述 除去情懷需求，互聯網上有大量珍貴的開放資料，近年來深度學習如雨后春筍一般火熱起來，但機器學習很多時候并不是苦于我的模型是否建立得合適，我的參數是否調整得正確，而是苦于最初的起步階段：沒有數據。 作為收集數據的前置工作，有能力去寫一個簡單的或者復雜的爬蟲，對于我們來說依然非常重要。 ## 基于colly的單機爬蟲 <details> <summary>main.go</summary> ``` package main import ( "fmt" "regexp" "time" "github.com/gocolly/colly" ) var visited = map[string]bool{} func main() { // Instantiate default collector c := colly.NewCollector( colly.AllowedDomains("www.abcdefg.com"), colly.MaxDepth(1), ) // 我們認為匹配該模式的是該網站的詳情頁 detailRegex, _ := regexp.Compile(`/go/go\?p=\d+$`) // 匹配下面模式的是該網站的列表頁 listRegex, _ := regexp.Compile(`/t/\d+#\w+`) // 所有a標簽，上設置回調函數 c.OnHTML("a[href]", func(e *colly.HTMLElement) { link := e.Attr("href") // 已訪問過的詳情頁或列表頁，跳過 if visited[link] && (detailRegex.Match([]byte(link)) || listRegex.Match([]byte(link))) { return } // 既不是列表頁，也不是詳情頁 // 那么不是我們關心的內容，要跳過 if !detailRegex.Match([]byte(link)) && !listRegex.Match([]byte(link)) { println("not match", link) return } // 因為大多數網站有反爬蟲策略 // 所以爬蟲邏輯中應該有 sleep 邏輯以避免被封殺 time.Sleep(time.Second) println("match", link) visited[link] = true time.Sleep(time.Millisecond * 2) c.Visit(e.Request.AbsoluteURL(link)) }) err := c.Visit("https://www.abcdefg.com/go/go") if err != nil {fmt.Println(err)} } ``` </details> ## 分布式爬蟲 在很多場景下，速度是有意義的： 1. 對于價格戰期間的電商們來說，希望能夠在對手價格變動后第一時間獲取到其最新價格，再靠機器自動調整本家的商品價格。 2. 對于類似頭條之類的Feed流業務，信息的時效性也非常重要。如果我們慢吞吞地爬到的新聞是昨天的新聞，那對于用戶來說就沒有任何意義。  列表頁的html內容中會包含有所有詳情頁的鏈接。詳情頁的數量一般是列表頁的10到100倍，所以我們將這些詳情頁鏈接作為“任務”內容，通過消息隊列分發出去。 針對頁面爬取來說，在執行時是否偶爾會有重復其實不太重要，因為任務結果是冪等的（這里我們只爬頁面內容，不考慮評論部分）。 ### nats [nats](../../../%E6%B6%88%E6%81%AF%E4%B8%AD%E9%97%B4%E4%BB%B6/nats.md) ### 結合nats和colly結合 我們為每一個網站定制一個對應的collector，并設置相應的規則，比如abcdefg，hijklmn（虛構的），再用簡單的工廠方法來將該collector和其host對應起來，每個站點爬到列表頁之后，需要在當前程序中把所有鏈接解析出來，并把落地頁的URL發往消息隊列 <details> <summary>消息生產 main.go</summary> ``` package main import ( "fmt" "net/url" "github.com/gocolly/colly" ) var domain2Collector = map[string]*colly.Collector{} var nc *nats.Conn var maxDepth = 10 var natsURL = "nats://localhost:4222" func factory(urlStr string) *colly.Collector { u, _ := url.Parse(urlStr) return domain2Collector[u.Host] } func initABCDECollector() *colly.Collector { c := colly.NewCollector( colly.AllowedDomains("www.abcdefg.com"), colly.MaxDepth(maxDepth), ) c.OnResponse(func(resp *colly.Response) { // 做一些爬完之后的善后工作 // 比如頁面已爬完的確認存進 MySQL }) c.OnHTML("a[href]", func(e *colly.HTMLElement) { // 基本的反爬蟲策略 link := e.Attr("href") time.Sleep(time.Second * 2) // 正則 match 列表頁的話，就 visit if listRegex.Match([]byte(link)) { c.Visit(e.Request.AbsoluteURL(link)) } // 正則 match 落地頁的話，就發消息隊列 if detailRegex.Match([]byte(link)) { err = nc.Publish("tasks", []byte(link)) nc.Flush() } }) return c } func initHIJKLCollector() *colly.Collector { c := colly.NewCollector( colly.AllowedDomains("www.hijklmn.com"), colly.MaxDepth(maxDepth), ) c.OnHTML("a[href]", func(e *colly.HTMLElement) { }) return c } func init() { domain2Collector["www.abcdefg.com"] = initABCDECollector() domain2Collector["www.hijklmn.com"] = initHIJKLCollector() var err error nc, err = nats.Connect(natsURL) if err != nil {os.Exit(1)} } func main() { urls := []string{"https://www.abcdefg.com", "https://www.hijklmn.com"} for _, url := range urls { instance := factory(url) instance.Visit(url) } } ``` </details> <details> <summary>消息消費 main.go</summary> ``` package main import ( "fmt" "net/url" "github.com/gocolly/colly" ) var domain2Collector = map[string]*colly.Collector{} var nc *nats.Conn var maxDepth = 10 var natsURL = "nats://localhost:4222" func factory(urlStr string) *colly.Collector { u, _ := url.Parse(urlStr) return domain2Collector[u.Host] } func initV2exCollector() *colly.Collector { c := colly.NewCollector( colly.AllowedDomains("www.abcdefg.com"), colly.MaxDepth(maxDepth), ) return c } func initV2fxCollector() *colly.Collector { c := colly.NewCollector( colly.AllowedDomains("www.hijklmn.com"), colly.MaxDepth(maxDepth), ) return c } func init() { domain2Collector["www.abcdefg.com"] = initV2exCollector() domain2Collector["www.hijklmn.com"] = initV2fxCollector() var err error nc, err = nats.Connect(natsURL) if err != nil {os.Exit(1)} } func startConsumer() { nc, err := nats.Connect(nats.DefaultURL) if err != nil {return} sub, err := nc.QueueSubscribeSync("tasks", "workers") if err != nil {return} var msg *nats.Msg for { msg, err = sub.NextMsg(time.Hour * 10000) if err != nil {break} urlStr := string(msg.Data) ins := factory(urlStr) // 因為最下游拿到的一定是對應網站的落地頁 // 所以不用進行多余的判斷了，直接爬內容即可 ins.Visit(urlStr) // 防止被封殺 time.Sleep(time.Second) } } func main() { startConsumer() } ``` </details>