-1

我正在编写一个自定义脚本以从 RackSpace cloudfiles 容器中获取数据，并列出给定容器中的所有文件（容器有大约 1 亿个文件），我一直致力于并行化代码，目前卡住了。


// function to read data from channel and display
// currently just displaying, but there will be allot of processing done on this data
func extractObjectItemsFromList(objListChan <-chan []string) {
    fmt.Println("ExtractObjectItemsFromList")
    for _, c := range <-objListChan {
        fmt.Println(urlPrefix, c, "\t", count)
    }
}


func main()

// fetching data using flags
ao := gophercloud.AuthOptions{
    Username: *userName,
    APIKey:   *apiKey,
}

provider, err := rackspace.AuthenticatedClient(ao)
client, err := rackspace.NewObjectStorageV1(provider,gophercloud.EndpointOpts{
    Region: *region,
})
if err != nil {
    logFatal(err)
}

// We have the option of filtering objects by their attributes
opts := &objects.ListOpts{
    Full:   true,
    Prefix: *prefix,
}

var objectListChan = make(chan []string)
go extractObjectItemsFromList(objectListChan)

// Retrieve a pager (i.e. a paginated collection)
pager := objects.List(client, *containerName, opts)


// Not working
// By default EachPage contains 10000 records
// Define an anonymous function to be executed on each page's iteration
lerr := pager.EachPage(func(page pagination.Page) (bool, error) {       // Get a slice of objects.Object structs
    objectList, err := objects.ExtractNames(page)
    if err != nil {
        logFatal(err)
    }
    for _, o := range objectList {
        _ = o
    }
    objectListChan <- objectList
    return true, nil
})
if lerr != nil {
    logFatal(lerr)
}
//---------------------------------------------------
//       below code is working
//---------------------------------------------------

// working, but only works inside the loop, this keeps on fetching new pages and showing new records, 10000 per page
// By default EachPage contains 10000 records
// Define an anonymous function to be executed on each page's iteration
lerr := pager.EachPage(func(page pagination.Page) (bool, error) {       // Get a slice of objects.Object structs
    objectList, err := objects.ExtractNames(page)
    if err != nil {
        logFatal(err)
    }
    for _, o := range objectList {
        fmt.Println(o)
    }
    return true, nil
})
if lerr != nil {
    logFatal(lerr)
}

显示前 10000 条记录，但随后卡住了，没有任何反应。如果我不使用通道而只运行普通循环，它工作得非常好，这会破坏并行化的目的。

4

1 回答 1

0 
for _, c := range <-objListChan {
    fmt.Println(urlPrefix, c, "\t", count)
}

您的异步工作人员从频道中弹出一个列表，对其进行迭代并退出。您需要有两个循环：一个读取通道 ( range objListChan)，另一个 - 读取（刚刚检索到的）对象列表。

于 2019-06-28T21:54:19.120 回答