Как заставить goroutines работать с анонимными функциями, возвращающими значение в цикле - PullRequest
Новая
       29

Как заставить goroutines работать с анонимными функциями, возвращающими значение в цикле

0 голосов
/ 29 июня 2019

Я работаю над пользовательским сценарием для извлечения данных из контейнера облачных файлов RackSpace и составления списка всех файлов в данном контейнере (контейнер содержит около 100 миллионов файлов), и я работаю над распараллеливанием кода и в настоящее время застрял.


// 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 записей, но затем они застряли, и ничего не происходит.Если я не использую канал и просто запускаю простой цикл, он прекрасно работает, что убивает цель распараллеливания.

1 Ответ

0 голосов
/ 29 июня 2019 
for _, c := range <-objListChan {
    fmt.Println(urlPrefix, c, "\t", count)
}

Ваш асинхронный рабочий извлекает один список из канала, повторяет его и завершает работу. Вам нужно иметь два цикла: один читает канал (range objListChan), другой - читает (только что извлеченный) список объектов.

...