optimization get personId function

sunty

2019-09-25 9969f8072b23fd8d03cb38445a81ae881bd01ee2

optimization get personId function

 EsApi.go

@@ -286,6 +286,7 @@

//获取当前节点抓拍库所有人员ID
func GetAllLocalVideopersonsId(compareArgs  protomsg.CompareArgs,indexName string, serverIp string, serverPort string) (capturetable []string) {
    ts := time.Now()
    queryStr := ""
    queryBody := compareArgs.InputValue
    //检索框
@@ -326,7 +327,6 @@

    //使用es底层机制处理分页


    analyServerId := compareArgs.AnalyServerId
    if analyServerId == "" {
        fmt.Println("no analyServerId")
@@ -334,13 +334,18 @@
    }
    analyServerFilterStr := "{\"term\":{\"analyServerId\":\"" + analyServerId + "\"}},"

    //请求头
    //首次请求头
    url := "http://" + serverIp + ":" + serverPort +
        "/" + indexName + "/_search?search_type=dfs_query_then_fetch"
        "/" + indexName + "/_search?search_type=dfs_query_then_fetch;scroll=1m"

    var lock sync.RWMutex
    var wg sync.WaitGroup

    for i := 0; i < 32; i++ {
    //请求体
    prama := "{" +
        "\"size\":\"100000000\"," +
            "\"slice\":{\"id\":" + strconv.Itoa(i) + ",\"max\":48}," +
            "\"size\":\"1000\"," +
        "\"query\":{\"bool\":{" + queryStr +
        "\"filter\":[" +
        cameraIdStr +
@@ -351,20 +356,86 @@
        "{\"range\":{\"picDate\":{\"from\":\"" + gteDate + "\",\"to\":\"" + lteDate + "\",\"include_lower\":true,\"include_upper\":true,\"boost\":1}}}]}}," +
        "\"_source\":[\"id\"]" +
        "}"
    fmt.Println(url)
    fmt.Println(prama)
        wg.Add(1)
        go func() {
            defer wg.Done()

            //fmt.Println(url)
            //fmt.Println(prama)
    buf, err := EsReq("POST", url,[]byte(prama))

    if err != nil {
        fmt.Println("http request videoUrlInfo info is err!")
                fmt.Println(len(capturetable))
        return
    }
    sources, err := Sourcelist(buf)

            sources, err := Sourcelistforscroll(buf)

    if err != nil {
                fmt.Println(len(capturetable))
        return
    }
    for _, source := range  sources{
            for _, source := range sources["sourcelist"].([]map[string]interface{}) {
        capturetable = append(capturetable, source["id"].(string))
    }

            scroll_id := sources["scroll_id"].(string)

            //scroll请求头
            scroll_url := "http://" + serverIp + ":" + serverPort + "/_search/scroll"

            for {
                var tmpList []string
                next_scroll_id := ""
                if next_scroll_id != "" {
                    scroll_id = next_scroll_id
                }
                jsonDSL := `{
            "scroll": "1m", 
            "scroll_id" : "` + scroll_id + `"
        }`
                //fmt.Println(scroll_url)
                //fmt.Println(jsonDSL)
                buf, err := EsReq("POST", scroll_url, []byte(jsonDSL))

                if err != nil {
                    fmt.Println("lenth1: ", len(capturetable))
                    return
                }
                nextSources, err := Sourcelistforscroll(buf)

                if nextSources == nil {
                    return
                }

                nextM := nextSources["sourcelist"].([]map[string]interface{})
                //fmt.Println("id",nextSources)
                if nextM == nil || len(nextM) == 0 {
                    //fmt.Println("lenth: ", len(capturetable))
                    return
                }
                //fmt.Println("id")
                for _, source := range nextM {
                    tmpList = append(tmpList, source["id"].(string))
                }
                //fmt.Println("tmpList: ", len(tmpList))
                lock.Lock()
                capturetable = append(capturetable, tmpList...)
                lock.Unlock()

                next_scroll_id = nextSources["scroll_id"].(string)
            }


            fmt.Println(len(capturetable))

        }()
    }
    wg.Wait()

    //fmt.Println("lenth_all: ", len(capturetable))
    //fmt.Println("耗时：", time.Since(ts))
    return capturetable
}


 EsClient.go

@@ -359,6 +359,45 @@
    }
    return sources,nil
}
//slice scroll 解析工具函数
func Sourcelistforscroll(buf []byte)(datasource map[string]interface{}, err error){
    var data = make(map[string]interface{})
    var info interface{}
    json.Unmarshal(buf, &info)
    out, ok := info.(map[string]interface{})
    if !ok {
        return nil, errors.New("http response interface can not change map[string]interface{}")
    }
    scroll_id, ok := out["_scroll_id"].(string)
    if !ok {
        return nil, errors.New("first hits change error!")
    }

    //fmt.Println("middle: ",scroll_id)
    middle, ok := out["hits"].(map[string]interface{})
    //fmt.Println("middle: ",out)
    if !ok {
        return nil, errors.New("first hits change error!")
    }
    var sources  = make([]map[string]interface{},0)
    for _, in := range middle["hits"].([]interface{}){
        tmpbuf, ok := in.(map[string]interface{})
        if !ok {
            fmt.Println("change to source error!")
            continue
        }
        source, ok := tmpbuf["_source"].(map[string]interface{})
        if !ok {
            fmt.Println("change _source error!")
            continue
        }
        sources  = append(sources, source )
    }

    data["sourcelist"] = sources
    data["scroll_id"] = scroll_id
    return data,nil
}

func EsReq(method string, url string, parama []byte) (buf []byte, err error) {
    defer elapsed("page")()