Как разделить один экземпляр HTTP-запроса между двумя горутинами?

Question

У меня есть код, который делает 3 запроса для заполнения 3 переменных. Два запроса одинаковы. Я хочу разделить один HTTP-запрос между двумя разными функциями (в реальном мире эти функции разделены на два разных модуля).

Позвольте мне описать возникшую у меня проблему на гораздо более простом примере, чем в реальном мире.

На данный момент у меня есть следующая основная функция и структура данных Post:

type Post struct {
    ID          int    `json:"id"`
    Title       string `json:"title"`
    UserID      int    `json:"userId"`
    isCompleted bool   `json:"completed"`
}

func main() {
    var wg sync.WaitGroup

    fmt.Println("Hello, world.")

    wg.Add(3)

    var firstPostID int
    var secondPostID int
    var secondPostName string

    go func() {
        firstPostID = getFirstPostID()
        defer wg.Done()
    }()

    go func() {
        secondPostID = getSecondPostID()
        defer wg.Done()
    }()

    go func() {
        secondPostName = getSecondPostName()
        defer wg.Done()
    }()

    wg.Wait()

    fmt.Println("first post id is", firstPostID)
    fmt.Println("second post id is", secondPostID)
    fmt.Println("second post title is", secondPostName)
}

Есть три горутины, поэтому у меня есть 3 одновременных запроса, я все синхронизирую, используя sync.Workgroup. Следующий код - это реализация запросов:

func makeRequest(url string) Post {
    resp, err := http.Get(url)
    if err != nil {
        // handle error
    }
    defer resp.Body.Close()
    body, err := ioutil.ReadAll(resp.Body)
    var post Post

    json.Unmarshal(body, &post)

    return post
}

func makeFirstPostRequest() Post {
    return makeRequest("https://jsonplaceholder.typicode.com/todos/1")
}

func makeSecondPostRequest() Post {
    return makeRequest("https://jsonplaceholder.typicode.com/todos/2")
}

Вот реализация функций, которые извлекают необходимую информацию из полученных сообщений:

func getFirstPostID() int {
    var result = makeFirstPostRequest()
    return result.ID
}

func getSecondPostID() int {
    var result = makeSecondPostRequest()

    return result.ID
}

func getSecondPostName() string {
    var result = makeSecondPostRequest()

    return result.Title
}

Итак, на данный момент у меня есть 3 одновременных запроса, это отлично работает. Проблема в том, что мне не нужны два абсолютно одинаковых отдельных HTTP-запроса для получения второго сообщения. Одного было бы достаточно. Итак, я хочу достичь двух одновременных запросов на сообщение 1 и сообщение 2. Я хочу, чтобы второй вызовmakeSecondPostRequest не создавать новый HTTP-запрос, а делиться существующим (который был отправлен первым вызовом).

Как я могу этого добиться?

Примечание. В следующем коде показано, как это можно сделать, например, с помощью JavaScript.

let promise = null;
function makeRequest() {
    if (promise) {
        return promise;
    }

    return promise = fetch('https://jsonplaceholder.typicode.com/todos/1')
      .then(result => result.json())
      // clean up cache variable, so any next request in the future will be performed again
      .finally(() => (promise = null))

}

function main() {
    makeRequest().then((post) => {
        console.log(post.id);
    });
    makeRequest().then((post) => {
        console.log(post.title);
    });
}

main();

Answer 1

Хотя вы можете составить что-то вроде обещаний, в этом случае в этом нет необходимости.

Ваш код написан в процедурном стиле. Вы написали очень специфические функции, которые извлекают изPostа остальные выбросить. Вместо этого держитеPost вместе.

package main

import(
    "fmt"
    "encoding/json"
    "net/http"
    "sync"
)

type Post struct {
    ID          int    `json:"id"`
    Title       string `json:"title"`
    UserID      int    `json:"userId"`
    isCompleted bool   `json:"completed"`
}

func fetchPost(id int) Post {
    url := fmt.Sprintf("https://jsonplaceholder.typicode.com/todos/%v", id)
    resp, err := http.Get(url)
    if err != nil {
        panic("HTTP error")
    }
    defer resp.Body.Close()

    // It's more efficient to let json Decoder handle the IO.
    var post Post
    decoder := json.NewDecoder(resp.Body)
    err = decoder.Decode(&post)
    if err != nil {
        panic("Decoding error")
    }

    return post
}

func main() {
    var wg sync.WaitGroup

    wg.Add(2)

    var firstPost Post
    var secondPost Post

    go func() {
        firstPost = fetchPost(1)
        defer wg.Done()
    }()

    go func() {
        secondPost = fetchPost(2)
        defer wg.Done()
    }()

    wg.Wait()

    fmt.Println("First post ID is", firstPost.ID)
    fmt.Println("Second post ID is", secondPost.ID)
    fmt.Println("Second post title is", secondPost.Title)
}

Теперь вместо кеширования ответов вы можете кешировать сообщения. Мы можем сделать это, добавив PostManager для обработки выборки и кеширования сообщений.

Обратите внимание, что нормальный mapнебезопасно для одновременного использования, поэтому мы используем sync.Map для нашего кеша.

type PostManager struct {
    sync.Map
}

func (pc *PostManager) Fetch(id int) Post {
    post, ok := pc.Load(id)
    if ok {
        return post.(Post)
    }
    post = pc.fetchPost(id)
    pc.Store(id, post)

    return post.(Post)
}

func (pc *PostManager) fetchPost(id int) Post {    
    url := fmt.Sprintf("https://jsonplaceholder.typicode.com/todos/%v", id)
    resp, err := http.Get(url)
    if err != nil {
        panic("HTTP error")
    }
    defer resp.Body.Close()

    // It's more efficient to let json Decoder handle the IO.
    var post Post
    decoder := json.NewDecoder(resp.Body)
    err = decoder.Decode(&post)
    if err != nil {
        panic("Decoding error")
    }

    return post
}

PostManager методы должны принимать указатель-получатель, чтобы избежать копирования мьютекса внутри sync.Map.

И вместо того, чтобы получать сообщения напрямую, мы используем PostManager.

func main() {
    var postManager PostManager

    var wg sync.WaitGroup

    wg.Add(2)

    var firstPost Post
    var secondPost Post

    go func() {
        firstPost = postManager.Fetch(1)
        defer wg.Done()
    }()

    go func() {
        secondPost = postManager.Fetch(2)
        defer wg.Done()
    }()

    wg.Wait()

    fmt.Println("First post ID is", firstPost.ID)
    fmt.Println("Second post ID is", secondPost.ID)
    fmt.Println("Second post title is", secondPost.Title)
}

Кэширование PostManager можно было бы улучшить за счет использования условных запросов для проверки того, изменилась ли кэшированная публикация или нет.

Его блокировка также может быть улучшена, поскольку написано, что можно одновременно получить одну и ту же публикацию. Мы можем исправить это, используя singleflight разрешить только один звонок fetchPost с заданным идентификатором происходить одновременно.

type PostManager struct {
    group singleflight.Group
    cached sync.Map
}

func (pc *PostManager) Fetch(id int) Post {
    post,ok := pc.cached.Load(id)
    if !ok {
        // Multiple calls with the same key at the same time will only run the code once, but all calls get the result.
        post, _, _ = pc.group.Do(strconv.Itoa(id), func() (interface{}, error) {
            post := pc.fetchPost(id)
            pc.cached.Store(id, post)
            return post, nil
        })
    }
    return post.(Post)
}