Panov_Alexey 18 июл 2022 в 11:36

Go sync/singleflight: устранение дублирования идентичных запросов

17 мин

5.9K

Go*Параллельное программирование*Программирование*

Из песочницы

Перевод

Автор оригинала: Kale Blankenship

Вступление от переводчика

На одном из проектов довелось столкнуться с чрезмерной нагрузкой на базу данных при холодном старте проекта, а порой и при устаревании кэша. Анализ проблемы показал, что из тысяч популярных запросов несколько вызывались на порядок чаще других. Настолько часто, что за время построения кэша после первого пришедшего запроса успевало прилететь еще с десяток запросов и все они параллельно нещадно грузили базу данных.

Проблему в тот раз удалось решить не самым красивым образом. Поэтому, когда намедни столкнулся с похожим кейсом уже на Go-стеке, первым делом отправился в Google, чтобы не наступать на прежние грабли. К своей радости обнаружил подходящий пакет singleflight (ссылка на оф документацию), предоставляющий механизм подавления повторяющихся вызовов.

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

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

Сталкивались ли Вы когда-нибудь с ситуацией, что в ваше приложение поступало несколько одновременных запросов к ресурсу, требующему вызова дорогостоящей функции (например, чтение файла, доступ к сети, выполнение вычисления) для обработки запроса?
Если Ваш ответ «да», Вам поможет sync/singleflight.

Сам по себе пакет singleflight не нов. Он уже давно (прим. переводчика: с 5 октября 2016) находится в стандартной библиотеке Go в качестве внутреннего пакета (для непосвященных: импорт пути, содержащего элемент «internal», запрещен, если код импорта находится за пределами дерева, имеющего корень в родительском элементе «internal» каталога).

Как итог пакет был скопирован в различные репозитории, не синхронизированные ни между собой, ни со стандартной версией, что не позволяло использовать удобство простого импорта пакета golang.org/x.

Достаточно разговоров, перейдем к конкретным примерам. Наиболее очевидное использование может заключаться в гарантировании отсутствия дублирования сетевых запросов при нескольких одновременных поисках DNS для одного и того же доменного имени.

Но не буду заострять внимание на этом примере, т.к. стандартная библиотека уже умеет это делать; именно поэтому был создан singleflight. Вместо этого рассмотрим пример HTTP-сервера, которому необходимо совершать запросы к внешнему API.

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"
)

func main() {
	// Create an HTTP handler
	http.HandleFunc("/github", func(w http.ResponseWriter, r *http.Request) {
		// Retrieve GitHub's API status
		status, err := githubStatus()
		if err != nil {
			// Send an internal error response if we were unable to retrieve the status.
			// note: not a great idea to raw errors to a client, but this is just a demonstration.
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		// Log the status so we can see what's happening from the server's perspective.
		log.Printf("/github handler requst: status %q", status)

		// Write the response to the client.
		fmt.Fprintf(w, "GitHub Status: %q", status)
	})

	http.ListenAndServe("127.0.0.1:8080", nil)
}

// githubStatus retrieves GitHub's API status
func githubStatus() (string, error) {
	// Log the start and end of the function so we can see how many times it's called.
	log.Println("Making request to GitHub API")
	defer log.Println("Request to GitHub API Complete") // The defer causes this to be logged after the function's return statement.

	// Atrificially delay this function to emulate a long running task
	time.Sleep(1 * time.Second)

	// Make a request to the GitHub Status API
	resp, err := http.Get("https://status.github.com/api/status.json")
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	// Check that we got a good response.
	if resp.StatusCode != 200 {
		return "", fmt.Errorf("github response: %s", resp.Status)
	}

	// Anonymous struct to extract the status from the response
	r := struct{ Status string }{}

	// Decode the JSON response
	err = json.NewDecoder(resp.Body).Decode(&r)

	return r.Status, err
}

В данном примере видим простой обработчик, совершающий запрос к GitHub и возвращающий статус API нашему клиенту в виде текстовой строки.

Теперь сделаем несколько одновременных запросов и посмотрим, что получится. Есть много способов сделать это, но я буду использовать vegeta, инструмент нагрузочного тестирования HTTP, написанный на Go.

# echo “GET http://localhost:8080/github" | vegeta attack -duration=1s -rate=10 | vegeta report
Requests [total, rate] 10, 11.11
Duration [total, attack, wait] 2.011072002s, 899.999ms, 1.111073002s
Latencies [mean, 50, 95, 99, max] 1.269280654s, 1.222384648s, 1.445123078s, 1.445123078s, 1.526538858s
Bytes In [total, mean] 210, 21.00
Bytes Out [total, mean] 0, 0.00
Success [ratio] 100.00%
Status Codes [code:count] 200:10
Error Set:

Введенная команда отправляет с помощью vegeta 10 GET запросов в секунду на наш эндпоинт в течение 1 секунды.

Из логов видно, что функция githubStatus() вызывалась 10 раз, по одному разу для каждого запроса.

# go run e1_without_singleflight.go
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:30 Making request to GitHub API
2016/10/11 12:36:31 Making request to GitHub API
2016/10/11 12:36:31 Request to GitHub API Complete
2016/10/11 12:36:31 /github handler requst: status “good”
2016/10/11 12:36:31 Request to GitHub API Complete
2016/10/11 12:36:31 /github handler requst: status “good”
2016/10/11 12:36:31 Request to GitHub API Complete
2016/10/11 12:36:31 /github handler requst: status “good”
2016/10/11 12:36:31 Request to GitHub API Complete
2016/10/11 12:36:31 /github handler requst: status “good”
2016/10/11 12:36:31 Request to GitHub API Complete
2016/10/11 12:36:31 /github handler requst: status “good”
2016/10/11 12:36:31 Request to GitHub API Complete
2016/10/11 12:36:31 /github handler requst: status “good”
2016/10/11 12:36:31 Request to GitHub API Complete
2016/10/11 12:36:31 /github handler requst: status “good”
2016/10/11 12:36:32 Request to GitHub API Complete
2016/10/11 12:36:32 /github handler requst: status “good”
2016/10/11 12:36:32 Request to GitHub API Complete
2016/10/11 12:36:32 /github handler requst: status “good”
2016/10/11 12:36:32 Request to GitHub API Complete
2016/10/11 12:36:32 /github handler requst: status “good”

Давайте подумаем, что здесь происходит. Почти одновременно поступает 10 запросов, после чего наш код инициирует 10 отдельных HTTP-запросов для получения одной и той же информации. Как результат, наше приложение должно выполнить много лишней работы. Мы перегружаем канал связи и совершаем избыточные сетевые запросы к API GitHub.

Посмотрим, что пакет singleflight может сделать для нас.

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"

	"golang.org/x/sync/singleflight"
)

func main() {
	// We need a Group to use singleflight.
	var requestGroup singleflight.Group

	http.HandleFunc("/github", func(w http.ResponseWriter, r *http.Request) {
		// This time we'll wrap the githubStatus() call with singleflight's Group.Do()
		// Do takes a key (more on this later) and a function that returns a interface{} and an error.
		v, err, shared := requestGroup.Do("github", func() (interface{}, error) {
			// githubStatus() returns string, error, which statifies interface{}, error, so we can return the result directly.
			return githubStatus()
		})
		// Do returns an interface{}, error, and a bool which indicates whether multiple calls to the function shared the same result.

		// Check the error, as before.
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		// We know that v will be a string, so we'll use a type assertion.
		status := v.(string)

		// Update the log statement so we can see if the results were shared.
		log.Printf("/github handler requst: status %q, shared result %t", status, shared)

		fmt.Fprintf(w, "GitHub Status: %q", status)
	})

	http.ListenAndServe("127.0.0.1:8080", nil)
}

// githubStatus retrieves GitHub's API status
func githubStatus() (string, error) {
	// No changes made to this function other than removing the comments for brevity.
	log.Println("Making request to GitHub API")
	defer log.Println("Request to GitHub API Complete")

	time.Sleep(1 * time.Second)

	resp, err := http.Get("https://status.github.com/api/status.json")
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return "", fmt.Errorf("github response: %s", resp.Status)
	}

	r := struct{ Status string }{}

	err = json.NewDecoder(resp.Body).Decode(&r)

	return r.Status, err
}

Какой же результат vegeta поможет нам получить теперь?

# go run e2_with_singleflight.go
2016/10/11 13:02:49 Making request to GitHub API
2016/10/11 13:02:51 Request to GitHub API Complete
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true
2016/10/11 13:02:51 /github handler requst: status “good”, shared result true

Гораздо лучше! На этот раз мы сделали всего один сетевой запрос к GitHub, а все остальные запросы вернули готовый результат первого. Важно обратить внимание на тот факт, что singleflight помогает избежать дублирования только для одновременных запросов - он не хранит в кэше результат после завершения вызова функции.

Мы можем убедиться в этом, запустив vegeta на 2 секунды (напомню, что в функции githubStatus() есть искусственная задержка в 1 секунду).

# echo “GET http://localhost:8080/github" | vegeta attack -duration=2s -rate=10 | vegeta report
Requests [total, rate] 20, 10.53
Duration [total, attack, wait] 2.609915466s, 1.899999924s, 709.915542ms
Latencies [mean, 50, 95, 99, max] 807.15366ms, 809.915542ms, 1.373483043s, 1.373483043s, 1.473610875s
Bytes In [total, mean] 420, 21.00
Bytes Out [total, mean] 0, 0.00
Success [ratio] 100.00%
Status Codes [code:count] 200:20
Error Set:
2016/10/11 13:17:07 Making request to GitHub API
2016/10/11 13:17:08 Request to GitHub API Complete
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 /github handler requst: status “good”, shared result true
2016/10/11 13:17:08 Making request to GitHub API
2016/10/11 13:17:10 Request to GitHub API Complete
2016/10/11 13:17:10 /github handler requst: status “good”, shared result true
2016/10/11 13:17:10 /github handler requst: status “good”, shared result true
2016/10/11 13:17:10 /github handler requst: status “good”, shared result true
2016/10/11 13:17:10 /github handler requst: status “good”, shared result true
2016/10/11 13:17:10 /github handler requst: status “good”, shared result true

Как мы видим, после завершения первого запроса к API GitHub отправляется второй. Отметим еще раз: singleflight предназначен для предотвращения множественных запросов в один момент времени, а не для кэширования.

Использование ключей

Мы пропустили тему использования ключей. Ключ используется, чтобы различать функции, которые мы хотим разделить. Чтобы разобраться лучше, давайте добавим в наше приложение эндпоинт получения статуса BitBucket API.

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"

	"golang.org/x/sync/singleflight"
)

func main() {
	var requestGroup singleflight.Group

	http.HandleFunc("/github", func(w http.ResponseWriter, r *http.Request) {
		v, err, shared := requestGroup.Do("github", func() (interface{}, error) {
			return githubStatus()
		})
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		status := v.(string)

		log.Printf("/github handler requst: status %q, shared result %t", status, shared)

		fmt.Fprintf(w, "GitHub Status: %q", status)
	})

	http.HandleFunc("/bitbucket", func(w http.ResponseWriter, r *http.Request) {
		// We can use the same singleflight.Group as long as we use a different key
		v, err, shared := requestGroup.Do("bitbucket", func() (interface{}, error) {
			return bitbucketStatus()
		})
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		status := v.(string)

		log.Printf("/bitbucket handler requst: status %q, shared result %t", status, shared)

		fmt.Fprintf(w, "BitBucket Status: %q", status)
	})

	http.ListenAndServe("127.0.0.1:8080", nil)
}

// githubStatus retrieves GitHub's API status
func githubStatus() (string, error) {
	log.Println("Making request to GitHub API")
	defer log.Println("Request to GitHub API Complete")

	time.Sleep(1 * time.Second)

	resp, err := http.Get("https://status.github.com/api/status.json")
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return "", fmt.Errorf("github response: %s", resp.Status)
	}

	r := struct{ Status string }{}

	err = json.NewDecoder(resp.Body).Decode(&r)

	return r.Status, err
}

// bitbucketStatus retrieves BitBucket's API status
func bitbucketStatus() (string, error) {
	log.Println("Making request to BitBucket API")
	defer log.Println("Request to BitBucket API Complete")

	time.Sleep(1 * time.Second)

	resp, err := http.Get("https://status.bitbucket.org/api/v2/status.json")
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	
	if resp.StatusCode != 200 {
		return "", fmt.Errorf("github response: %s", resp.Status)
	}

	r := struct{ Status struct{ Description string } }{}

	err = json.NewDecoder(resp.Body).Decode(&r)

	return r.Status.Description, err
}

Если не считать небольших различий в структуре JSON, хендлеры GitHub и BitBucket практически идентичны. Как указано в коде, другой ключ используется, чтобы отличать функции при использовании singleflight.Group.

echo “GET http://localhost:8080/github\nGET http://localhost:8080/bitbucket" | vegeta attack -duration=1s -rate=10 | vegeta report
Requests [total, rate] 10, 11.11
Duration [total, attack, wait] 1.554732379s, 899.999947ms, 654.732432ms
Latencies [mean, 50, 95, 99, max] 1.069608701s, 984.508873ms, 1.384495166s, 1.384495166s, 1.5546515s
Bytes In [total, mean] 320, 32.00
Bytes Out [total, mean] 0, 0.00
Success [ratio] 100.00%
Status Codes [code:count] 200:10
Error Set:
# go run e3_bitbucket.go
2016/10/11 13:27:48 Making request to GitHub API
2016/10/11 13:27:48 Making request to BitBucket API
2016/10/11 13:27:49 Request to BitBucket API Complete
2016/10/11 13:27:49 /bitbucket handler requst: status “All Systems Operational”, shared result true
2016/10/11 13:27:49 /bitbucket handler requst: status “All Systems Operational”, shared result true
2016/10/11 13:27:49 /bitbucket handler requst: status “All Systems Operational”, shared result true
2016/10/11 13:27:49 /bitbucket handler requst: status “All Systems Operational”, shared result true
2016/10/11 13:27:49 /bitbucket handler requst: status “All Systems Operational”, shared result true
2016/10/11 13:27:49 Request to GitHub API Complete
2016/10/11 13:27:49 /github handler requst: status “good”, shared result true
2016/10/11 13:27:49 /github handler requst: status “good”, shared result true
2016/10/11 13:27:49 /github handler requst: status “good”, shared result true
2016/10/11 13:27:49 /github handler requst: status “good”, shared result true
2016/10/11 13:27:49 /github handler requst: status “good”, shared result true

Наглядно видно, что функции githubStatus() и bitbucketStatus() были вызваны по одному разу, а возвращаемые строки состояния явно различаются.

Ради интереса посмотрим, что произойдет, если мы “случайно” воспользуемся одним и тем же ключом.


package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"

	"golang.org/x/sync/singleflight"
)

func main() {
	var requestGroup singleflight.Group

	http.HandleFunc("/github", func(w http.ResponseWriter, r *http.Request) {
		v, err, shared := requestGroup.Do("github", func() (interface{}, error) {
			return githubStatus()
		})
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		status := v.(string)

		log.Printf("/github handler requst: status %q, shared result %t", status, shared)

		fmt.Fprintf(w, "GitHub Status: %q", status)
	})

	http.HandleFunc("/bitbucket", func(w http.ResponseWriter, r *http.Request) {
		// Oops! Same key!
		v, err, shared := requestGroup.Do("github", func() (interface{}, error) {
			return bitbucketStatus()
		})
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		status := v.(string)

		log.Printf("/bitbucket handler requst: status %q, shared result %t", status, shared)

		fmt.Fprintf(w, "BitBucket Status: %q", status)
	})

	http.ListenAndServe("127.0.0.1:8080", nil)
}

// githubStatus retrieves GitHub's API status
func githubStatus() (string, error) {
	log.Println("Making request to GitHub API")
	defer log.Println("Request to GitHub API Complete")

	time.Sleep(1 * time.Second)

	resp, err := http.Get("https://status.github.com/api/status.json")
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return "", fmt.Errorf("github response: %s", resp.Status)
	}

	r := struct{ Status string }{}

	err = json.NewDecoder(resp.Body).Decode(&r)

	return r.Status, err
}

// bitbucketStatus retrieves BitBucket's API status
func bitbucketStatus() (string, error) {
	log.Println("Making request to BitBucket API")
	defer log.Println("Request to BitBucket API Complete")

	time.Sleep(1 * time.Second)

	resp, err := http.Get("https://status.bitbucket.org/api/v2/status.json")
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	
	if resp.StatusCode != 200 {
		return "", fmt.Errorf("github response: %s", resp.Status)
	}

	r := struct{ Status struct{ Description string } }{}

	err = json.NewDecoder(resp.Body).Decode(&r)

	return r.Status.Description, err
}

Результат:

# go run e4_same_key.go
2016/10/11 13:34:39 Making request to GitHub API
2016/10/11 13:34:40 Request to GitHub API Complete
2016/10/11 13:34:40 /github handler requst: status “good”, shared result true
2016/10/11 13:34:40 /github handler requst: status “good”, shared result true
2016/10/11 13:34:40 /github handler requst: status “good”, shared result true
2016/10/11 13:34:40 /bitbucket handler requst: status “good”, shared result true
2016/10/11 13:34:40 /bitbucket handler requst: status “good”, shared result true
2016/10/11 13:34:40 /bitbucket handler requst: status “good”, shared result true
2016/10/11 13:34:40 /github handler requst: status “good”, shared result true
2016/10/11 13:34:40 /github handler requst: status “good”, shared result true
2016/10/11 13:34:40 /bitbucket handler requst: status “good”, shared result true
2016/10/11 13:34:40 /bitbucket handler requst: status “good”, shared result true

Как и следовало ожидать, обработчик BitBucket возвращает результат для GitHub (или наоборот, в зависимости от того, какой обработчик выполняется первым).

Удаление ключа

Порой может потребоваться “забыть” ранее использованный ключ. Предположим, что мы хотим разделить ответ первого запроса лишь с запросами, которые придут в течение 250 мс после инициации первого запроса.

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"

	"golang.org/x/sync/singleflight"
)

func main() {
	var requestGroup singleflight.Group

	http.HandleFunc("/github", func(w http.ResponseWriter, r *http.Request) {
		v, err, shared := requestGroup.Do("github", func() (interface{}, error) {
			// Start a goroutine which deletes the "github" key after 250ms.
			go func() {
				time.Sleep(250 * time.Millisecond)
				log.Println("Deleting \"github\" key")
				requestGroup.Forget("github")
			}()

			return githubStatus()
		})
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}

		status := v.(string)

		log.Printf("/github handler requst: status %q, shared result %t", status, shared)

		fmt.Fprintf(w, "GitHub Status: %q", status)
	})

	http.ListenAndServe("127.0.0.1:8080", nil)
}

// githubStatus retrieves GitHub's API status
func githubStatus() (string, error) {
	log.Println("Making request to GitHub API")
	defer log.Println("Request to GitHub API Complete")

	time.Sleep(1 * time.Second)

	resp, err := http.Get("https://status.github.com/api/status.json")
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return "", fmt.Errorf("github response: %s", resp.Status)
	}

	r := struct{ Status string }{}

	err = json.NewDecoder(resp.Body).Decode(&r)

	return r.Status, err
}

Результат выполнения кода:

# echo “GET http://localhost:8080/github" | vegeta attack -duration=1s -rate=10 | vegeta report
Requests [total, rate] 10, 11.11
Duration [total, attack, wait] 2.007352469s, 899.999899ms, 1.10735257s
Latencies [mean, 50, 95, 99, max] 1.214591847s, 1.13383041s, 1.432634042s, 1.432634042s, 1.53263476s
Bytes In [total, mean] 210, 21.00
Bytes Out [total, mean] 0, 0.00
Success [ratio] 100.00%
Status Codes [code:count] 200:10
Error Set:
# go run e5_forget.go
2016/10/11 13:45:21 Making request to GitHub API
2016/10/11 13:45:21 Deleting “github” key
2016/10/11 13:45:21 Making request to GitHub API
2016/10/11 13:45:21 Deleting “github” key
2016/10/11 13:45:21 Making request to GitHub API
2016/10/11 13:45:22 Deleting “github” key
2016/10/11 13:45:22 Making request to GitHub API
2016/10/11 13:45:22 Deleting “github” key
2016/10/11 13:45:22 Request to GitHub API Complete
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:22 Request to GitHub API Complete
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:22 Request to GitHub API Complete
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:22 /github handler requst: status “good”, shared result true
2016/10/11 13:45:23 Request to GitHub API Complete
2016/10/11 13:45:23 /github handler requst: status “good”, shared result false

Судя по логам, было сделано четыре отдельных запроса, по одному через каждые 250 мс, хотя это может быть не столь очевидно на первый взгляд, как хотелось бы.

Обратите внимание, что удаление ключа оказывает влияние только на обработку запросов, поступивших после удаления ключа. При этом обработка уже выполняемых запросов продолжится без изменений.

Метод DoChan

В качестве альтернативы Do() пакет singleflight предоставляет метод DoChan().

Он похож на метод Do(), но возвращает канал, который получит результаты по мере готовности. Это может быть полезно, например, в случае реализации тайм-аута с помощью оператора select.

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"time"

	"golang.org/x/sync/singleflight"
)

func main() {
	var requestGroup singleflight.Group

	http.HandleFunc("/github", func(w http.ResponseWriter, r *http.Request) {
		// With DoChan we get a channel of singleflight.Result
		ch := requestGroup.DoChan("github", func() (interface{}, error) {
			return githubStatus()
		})

		// Create our timeout
		timeout := time.After(500 * time.Millisecond)

		var result singleflight.Result
		select {
		case <-timeout: // Timeout elapsed, send a timeout message (504)
			log.Println("/github handler timed out")
			http.Error(w, "github request timed out", http.StatusGatewayTimeout)
			return
		case result = <-ch: // Received result from channel
		}

		// singleflight.Result is the same three values as returned from Do(), but wrapped
		// in a struct. We can use the same logic as before, buy referencing the struct fields.

		if result.Err != nil {
			http.Error(w, result.Err.Error(), http.StatusInternalServerError)
			return
		}

		status := result.Val.(string)

		log.Printf("/github handler requst: status %q, shared result %t", status, result.Shared)

		fmt.Fprintf(w, "GitHub Status: %q", status)
	})

	http.ListenAndServe("127.0.0.1:8080", nil)
}

// githubStatus retrieves GitHub's API status
func githubStatus() (string, error) {
	log.Println("Making request to GitHub API")
	defer log.Println("Request to GitHub API Complete")

	time.Sleep(1 * time.Second)

	resp, err := http.Get("https://status.github.com/api/status.json")
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return "", fmt.Errorf("github response: %s", resp.Status)
	}

	r := struct{ Status string }{}

	err = json.NewDecoder(resp.Body).Decode(&r)

	return r.Status, err
}

Как уже было упомянуто, такой код всегда будет приводить к завершению по тайм-ауту. Рекомендую вам поэкспериментировать с этим примером, если Вы не знакомы с обработкой тайм-аута с помощью оператора select.

Результат выполнения кода:

# echo “GET http://localhost:8080/github" | vegeta attack -duration=1s -rate=10 | vegeta report
Requests [total, rate] 10, 11.11
Duration [total, attack, wait] 1.403848483s, 899.99994ms, 503.848543ms
Latencies [mean, 50, 95, 99, max] 506.630505ms, 506.02365ms, 509.411055ms, 509.411055ms, 511.554041ms
Bytes In [total, mean] 250, 25.00
Bytes Out [total, mean] 0, 0.00
Success [ratio] 0.00%
Status Codes [code:count] 504:10
Error Set:
504 Gateway Timeout

Vegeta ожидаемо сообщает о завершении с ошибкой 504 Gateway Time Out.

# go run e6_dochan.go
2016/10/11 14:14:32 Making request to GitHub API
2016/10/11 14:14:32 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:33 /github handler timed out
2016/10/11 14:14:34 Request to GitHub API Complete

Время ожидания всех запросов истекло.

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

Дополнительные ссылки

Пакет sync/singleflight

Теги:

Хабы:

Go sync/singleflight: устранение дублирования идентичных запросов

Вступление от переводчика

Использование ключей

Удаление ключа

Метод DoChan

Дополнительные ссылки

Публикации

Истории

Работа

Ближайшие события