Using rpc pool.

Author: ORACLE128G

crawler-distributed/main.go

@@ -1,24 +1,36 @@
 package main
 
 import (
+	"flag"
 	"polaris/crawler-distributed/config"
 	itemsaver "polaris/crawler-distributed/persist/client"
+	"polaris/crawler-distributed/pool"
 	worker "polaris/crawler-distributed/worker/client"
 	"polaris/crawler/engine"
 	"polaris/crawler/scheduler"
 	"polaris/crawler/zhenai/parser"
+	"strings"
 )
 
+var (
+	itemSaverHost = flag.String("itemsaver_host", "",
+		"itemsaver_host")
+
+	workerHosts = flag.String("worker_host", "",
+		"worker_host(comma separated)")
+)
 func main() {
 	// run condition:
 	// run itemsaver.go first
 	// then to run worker.go
-	itemChan, err := itemsaver.ItemSaver(config.ItemSaver0Host)
+	flag.Parse()
+	itemChan, err := itemsaver.ItemSaver(*itemSaverHost)
 	if err != nil {
 		panic(err)
 	}
 
-	processor, err := worker.CreateProcessor()
+	clientPool := pool.CreateClientPool(strings.Split(*workerHosts, ","))
+	processor, err := worker.CreateProcessor(clientPool)
 	if err != nil {
 		panic(err)
 	}
@@ -34,3 +46,4 @@ func main() {
 		Parser: engine.NewFuncParser(parser.ParseCityList, config.ParseCityList),
 	})
 }
+

crawler-distributed/persist/rpc.go

@@ -11,15 +11,14 @@ type ItemSaverService struct {
 	Index string
 }
 
-
 func (s *ItemSaverService) Save(item engine.Item, result *string) error {
-	const index  = "polaris"
+	const index = "polaris"
 	err := persist.Save(item, index)
 	if err == nil {
-		fmt.Println("saving item success")
+		fmt.Printf("saving item#%v \n", item.Url)
 		*result = "ok"
 	} else {
-		log.Error("saving item fail")
+		log.Error("saving item#%v fail", item)
 		*result = "failed"
 	}
 	return err

crawler-distributed/persist/server/itemsaver.go

@@ -1,7 +1,9 @@
 package main
 
 import (
-	"polaris/crawler-distributed/config"
+	"flag"
+	"fmt"
+	"log"
 	"polaris/crawler-distributed/persist"
 	"polaris/crawler-distributed/rpc"
 )
@@ -13,6 +15,14 @@ func serveRpc(host, index string) error {
 		})
 }
 
+var port = flag.Int("port", 0,
+	"The port for me to listen on")
+
 func main() {
-	serveRpc(config.ItemSaver0Host, "polaris")
+	flag.Parse()
+	if *port == 0 {
+		log.Printf("Must specify a port ")
+		return
+	}
+	serveRpc(fmt.Sprintf(":%d", *port), "polaris")
 }

crawler-distributed/pool/rpclientpool.go

@@ -0,0 +1,33 @@
+package pool
+
+import (
+	"log"
+	"net/rpc"
+	"polaris/crawler-distributed/rpc"
+)
+
+func CreateClientPool(hosts []string) chan *rpc.Client {
+
+	var clients []*rpc.Client
+	for _, h := range hosts{
+		client, err := rpcsupport.NewClient(h)
+		if err == nil {
+			clients = append(clients, client)
+			log.Printf("Connected to %s", h)
+		} else {
+			log.Printf("Error connecting to %s: %v",
+				h, err)
+		}
+	}
+
+
+	out := make(chan *rpc.Client)
+	go func() {
+		for {
+			for _, client := range clients {
+				out <- client
+			}
+		}
+	}()
+	return out
+}

crawler-distributed/rpc/rpc.go

@@ -13,7 +13,7 @@ func ServeRpc(host string, service interface{}) error {
 	if err != nil {
 		return err
 	}
-
+	log.Printf("Listen on %s", host)
 	for {
 		conn, err := listener.Accept()
 		if err != nil {

crawler-distributed/worker/client/worker.go

@@ -1,23 +1,21 @@
 package client
 
 import (
-	"fmt"
+	"net/rpc"
 	"polaris/crawler-distributed/config"
-	"polaris/crawler-distributed/rpc"
 	"polaris/crawler-distributed/worker"
 	"polaris/crawler/engine"
 )
 
-func CreateProcessor() (engine.Processor, error) {
-	client, err := rpcsupport.NewClient(fmt.Sprintf("%s", config.Worker0Host))
-	if err != nil {
-		return nil, err
-	}
+func CreateProcessor(
+	pool chan *rpc.Client) (engine.Processor, error) {
 
 	return func(request engine.Request) (engine.ParseResult, error) {
 		sReq := worker.SerializedRequest(request)
 		var sResult worker.ParseResult
-		err := client.Call(config.CrawlServiceRpc, sReq, &sResult)
+
+		c := <-pool
+		err := c.Call(config.CrawlServiceRpc, sReq, &sResult)
 		if err != nil {
 			return engine.ParseResult{}, err
 		}

crawler-distributed/worker/server/worker.go

@@ -1,14 +1,25 @@
 package main
 
 import (
-	"polaris/crawler-distributed/config"
+	"flag"
+	"fmt"
+	"log"
 	"polaris/crawler-distributed/rpc"
 	"polaris/crawler-distributed/worker"
 )
 
+var port = flag.Int("port", 0,
+	"The port for me to listen on")
+
 func main() {
-	err := rpcsupport.ServeRpc(config.Worker0Host, worker.CrawlService{})
-	if err != nil{
+	flag.Parse()
+	if *port == 0 {
+		log.Printf("Must specify a port ")
+		return
+	}
+	err := rpcsupport.ServeRpc(fmt.Sprintf(
+		":%d", *port), worker.CrawlService{})
+	if err != nil {
 		panic(err)
 	}
 }

crawler/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"polaris/crawler-distributed/config"
 	"polaris/crawler/engine"
 	"polaris/crawler/persist"
 	"polaris/crawler/scheduler"
@@ -19,16 +20,16 @@ func main() {
 		RequestProcessor: engine.Worker,
 	}
 
-	/*e.Run(engine.Request{
+	e.Run(engine.Request{
 		Url: seed,
 		Parser: engine.NewFuncParser(
 			parser.ParseCityList,
 			config.ParseCityList),
-	})*/
-	e.Run(engine.Request{
+	})
+	/*e.Run(engine.Request{
 		Url: shanghai,
 		Parser: engine.NewFuncParser(
 			parser.ParseCity,
 			"ParseCity"),
-	})
+	})*/
 }