Как реализовать многопоточность в go?

Question

[Алексей Маслов]

Допустим есть несколько телефонных станций, с которых необходимо одновременно собирать информацию и писать ее в какую-нибудь БД.

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

А хотелось бы, чтобы если первая и третья станция отвечают быстро - не ждать ответа второй и производить снова запрос к первой и третьей, и т.д.

Примерный код, для наглядности как это представлено сейчас:

func someRequest(phone string, ch chan bool)  {
	fmt.Println(phone)
	//time.Sleep(time.Second * 5) //Долгая операция. Для разных phone - разная по времени выполнения
	ch <- true
}

func main() {
	var phones = [2]string{"foo", "bar"}

	ch := make(chan bool) // канал
	for {
		fmt.Println("start request")

		for _, phone := range phones{
			go someRequest(phone, ch)
		}

		//ожидаем завершения всех каналов
		for i:=0 ; i < len(phones); i++{
			<-ch
		}
		fmt.Println("success")
	}
}

И дополнительно, когда получили данные со станции в гоурутине, перед записью в базу нужно ли пользоваться Мьютексами, чтобы не создать конфликтов или каких то коллизий при записи в БД?

Answer 1

[Александр Павлюк]

Можно создать канал для данных, которые надо писать в базу и по горутине на станцию, внутри каждой бесконечный цикл. Горутины постоянно читают каждая свою станцию и пишут в один канал данные, которые надо в БД заносить. Ещё одна горутина читает из этого канала через range и заносит данные в базу.

Comments

[Алексей Маслов]

Правильно ли я вас понял?

import (
	"fmt"
	"time"
)

//Запрос к станции
func someRequest(phone string, chData chan string)  {
	for {
		fmt.Println("start request", phone)
		time.Sleep(time.Second * 5) //Эмулируем задержку при выполнении
		chData <- "Data from stantion " + phone
	}
}

func main() {
	var phones = [5]string{"one", "two", "three", "four", "five"}

	var dataCh chan string //канал с данными со станции для записи в бд
	dataCh = make(chan string) // Инициализация канала

	go func(){ //Для записи в бд
		//Ожидаем данных в канал dataCh
		for {
			storeDbData := <- dataCh
			println("receive data from chanel and trying store to db", storeDbData)
		}
	}()

	for _, phone := range phones{
		go someRequest(phone, dataCh)
	}
}

Не совсем понял вашу фразу:

Ещё одна горутина читает из этого канала через range и заносит данные в базу.

Правильно ли я понимаю, что там где у меня анонимная функция (для записи в базу) будет всегда происходить ожидание пока канал не заполнится, затем он будет очищаться и снова ожидать.

Не повесят ли все эти бесконечные циклы систему? Чем лучше, на ваш взгляд, производить задержку между запросами в бесконечном цикле time.Sleep(time.Second)?

[Александр Павлюк]

Алексей Маслов, в данном коде у вас программа выйдёт из main и завершится сразу.
Я имел ввиду так:

package main

import (
	"fmt"
	"time"
)

// Получение данных со станции
func stationListener(phone string, chData chan string) {
	for {
		fmt.Println("start request", phone)
		time.Sleep(time.Second * 5) // Эмулируем ожидание ответа от станции
		chData <- "Data from station " + phone
	}
}

func main() {
	var phones = [5]string{"one", "two", "three", "four", "five"}

	dataCh := make(chan string) //канал с данными со станции для записи в бд

	for _, phone := range phones {
		go stationListener(phone, dataCh)
	}

	for storeDbData := range dataCh {
		println("receive data from chanel and trying store to db", storeDbData)
	}
}

Фраза про чтение через range имела ввиду for storeDbData := range dataCh {
Эта конструкция читает по-одной записи из канала, пока канал не будет закрыт.

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

Бесконечный цикл не повесит систему, если в цикле присутствует операция ввода-вывода. Ей может быть чтение/запись в канал, пауза через time, обращение к внешнему сервису, итд.

[Алексей Маслов]

Александр Павлюк, Огромное спасибо, с range dataCh сразу все прояснилось. Не знал про такой подход.

Если не сложно, не могли бы еще подсказать:
В каком месте лучше реализовать коннект к станции и рекконнект в случае обрыва соединения?
Допустим методе stationListener перед бесконечным циклом устанавливаем соединение. В бесконечном цикле какое-то время все работает, но потом вдруг обрыв сети.

Имеет ли смысл сделать цикл в цикле?
for {
коннект к станции
for{
ловим исключение - выкидываю из первого цикла
работа со станцией
отправляем сообщение в канал
}
дисконнект от станции (очистка ресурсов)
}

Или ловить исключение и посылать информацию о подключении в новый канал, и там производить реконнект к станции. (Но это мудрено как то)

Или из работающей гоурутины можно создать новую гоурутину (например во время обрыва соединения)?

И второй вопрос: на сколько большие данные можно передавать в каналы?

[Александр Павлюк]

Алексей Маслов, С двумя вложенными циклами выглядит лучше всего.

В канал можно пихать сколько угодно данных, хоть гигабайтный объект.

[Алексей Маслов]

Александр Павлюк, Еще раз большое спасибо!