package engine

import "fmt"

// 定义了一个
type ConcurrectEngine struct {
    Scheduler Scheduler
    WorkerCount int
}

type Scheduler interface{
    Submit(Request)
    ConfigureMasterWorkerChan(chan Request)
}

func (c *ConcurrectEngine) Run(seeds ...Request) {
    // 第一步: 做初始化工作
    in := make(chan Request)
    out := make(chan ParseResult)
    c.Scheduler.ConfigureMasterWorkerChan(in)

     创建工作者,从in输入管道中取出request进行处理,处理后放入到ParseResult的管道中
    for i := 0; i<c.WorkerCount; i++  {
         创建WorkerCount个工作者
        c.createWorker(in,out)
    }

     第二步: 把种子放到任务调度器中
    for _,seed := range seeds {
        c.Scheduler.Submit(seed)
    }

     第三步: 处理工作者线程返回的ParseResult,将items打印,requests再次添加到任务调度器中
    for {
        pr := <- out
         range pr.Items  {
            fmt.Printf(内容项: %v \n,item)
        }

         range pr.Req{
            c.Scheduler.Submit(p)
        }
    }
}

func (c *ConcurrectEngine) createWorker(in chan Request,1)"> chan ParseResult) {
    go func() {
         从in中取出request请求
        req := <- 
        parseResult,e := worker(req)
        if e != nil {
            return
        }
        out <- parseResult
    }()
}

 定义了一个并发的引擎结构,目的是,区别于之前写的单机版的Engine.
type ConcurrectEngine 
}

 定义了调度器的接口. 既然是调度器,就要有一个接受请求的方法.
type Scheduler {
     接收请求的方法
    Submit(Request)
     这里比较巧妙的地方是,其实调度器的chan类型和工作者输入的chan类型是同一个.
     调度器输出的request,就是工作者的输入request,这里让他们指向同一个地址.
    ConfigureMasterWorkerChan(chan Request)
}

func (c * range pr.Req{
            c.Scheduler.Submit(p)
        }
    }
}

package sechduler

import aaa/crawler/zhenai/engine

type SimpleScheduler  {
    workerChan chan engine.Request

}

 这里需要是一个地址拷贝,而不是值拷贝. 也就是workerChan和in使用的是同一个地址
func (s *SimpleScheduler) ConfigureMasterWorkerChan( chan engine.Request) {
    s.workerChan = 
}

 将请求添加到任务调度器
func (s *SimpleScheduler) Submit(r engine.Request) {
    go func() {
        s.workerChan <- r
    }()
}

func (c *ConcurrectEngine) createWorker(  {
             从in中取出request请求
            req := <- 
            parseResult,1)"> worker(req)
             nil {
                
            }
             parseResult
        }
    }()
}

type QueuedScheduler  {
    RequestChan chan engine.Request
     工作者channel,他的类型是request类型的chan. 工作者需要接受request类型的chan,然后进行工作
     多个工作者之间,也是一个chan
    WorkerChan chan chan engine.Request
}

func (q *QueuedScheduler) Submit(r engine.Request) {
    q.RequestChan <- r
}

func (q *QueuedScheduler) ConfigureMasterWorkerChan(chan engine.Request) {
    panic(implement me)
}

/**
 * 告诉我哪一个worker已经准备好,可以工作了
 */
func (q *QueuedScheduler) WorkerReady(worker chan engine.Request) {
    q.WorkerChan <- worker
}

func (q *QueuedScheduler) Run() {
    q.RequestChan = make(chan engine.Request)
    q.WorkerChan = make(chan chan engine.Request)
    go func() {
        var requestQ []engine.Request
         workerQ []chan engine.Request
         {
             requestActive engine.Request
             workerActive chan engine.Request
            if len(requestQ) > 0 && len(workerQ) > 0 {
                requestActive = requestQ[]
                workerActive = workerQ[]
            }

            selectcase r := <-q.RequestChan:
                 发送给worker
                requestQ = append(requestQ,r)
            case w := <-q.WorkerChan:
                 将下一个请求发送给worker
                workerQ = append(workerQ,w)
            case workerActive <- requestActive:
                    requestQ = requestQ[1:]
                    workerQ = workerQ[:]
            }
        }
    }()
}

"


    WorkChan chan {}
}

    ConfigureMasterWorkerChan(chan Request)
    WorkerReady(chan Request)
    Run()
}

func (c * 第一步: 做初始化工作
    out := make(chan ParseResult)
    c.Scheduler.Run()

     创建WorkerCount个工作者
        c.createWorker(c.Scheduler,requests再次添加到任务调度器中
    itemCount := 0
    内容项:  %d,%v \n
        }

         range pr.Req {
            c.Scheduler.Submit(p)
        }
    }
}

func (c *ConcurrectEngine) createWorker(sche Scheduler,1)"> chan ParseResult) {
    in := make(chan Request)
    go func() {
         告诉调度器,我已经准备好开始工作了
            sche.WorkerReady(in)
             parseResult
        }
    }()
}

    ConfigureMasterWorkerChan(chan Request)
    WorkerReady(chan Request)
    Run()
}

查看docker有哪些命令
docker

查看docker的版本信息
docker version

查看docker的基本信息
docker info

下载并运行elasticSearch
docker run -d -p9200:9200 daocloud.io/library/elasticsearch

这里的-p 后面跟的是端口号. 
第一个9200表示映射到物理机的端口是9200
第二个9200表示elasticSearch在虚拟机中运行的端口是9200

请求方式: POST
请求的url: localhost:9200/pachong/user/1

{
    _index": pachong_typeuser_id1_versionresultcreated_shards: {
        total2successful@R_404_159@
    },1)">": true
}

请求方式: GET
请求url: localhost:1

found_sourcenamelxlage12
    }
}

9200/pachong/user/_search
输入参数: 空

took97timed_outfalse5skippedhitsmax_score: [
            {
                2_score: {
                    ykk43
                }
            },{
                
                }
            }
        ]
    }
}

9200/pachong/user/_search?q=ykk

40.25811607
                }
            }
        ]
    }
}

func Save(item interface{}) (string 第一步: 创建一个elasticSearch client
     文档: https://godoc.org/gopkg.in/olivere/elastic.v5
    client,err := elastic.NewClient(elastic.SetURL(http://localhost:9200), sniff: 是用来维护客户端集群的状态的. 但是我们的集群不跑在本机上,而是跑在docker上.
         docker只有一个内网,内网我们看不见.所以没有办法维护状态,设置为false
        elastic.SetSniff())
    if err != nil {
        return "" 将item数据保存到的elasticSearch中
     elastic search 保存数据使用的Index.
     elasticSearch 数据中的三部分分别是 /index/type/id  . 对应数据库的/database/table/id
    response,err := client.Index().
        Index(dataing_profile).
        Type(zhenai).
        BodyJson(item).
        Do(context.Background())
     +v 表示打印结构体带上结构体的名称
     response.Id,nil

}

func TestSave(t *testing.T) {
    tests := [] {
        name 
        item model.Profile
    }{
        {冰靓晴雪已婚23白羊座15449北京10000-20000销售总监大学本科四川理财有房有车不要小孩 range tests {
        t.Run(tt.name,func(t *testing.T) {
            id,1)"> Save(tt.item)
             nil {
                panic(e)
            }

            client,e := elastic.NewClient(elastic.SetSniff())
             nil {
                panic(e)
            }

            resp,1)"> client.Get().
                Index().
                Type().
                Id(id).
                Do(context.Background())
             nil {
                panic(e)
            }

            fmt.Printf(%s",*resp.Source)

            user := model.Profile{}
            e = json.Unmarshal(*resp.Source,&user)
             nil {
                panic(e)
            }

            if user != tt.item {
                t.Errorf(error)
            }
        })
    }
}