Шаг 1. Ищем код эхо-сервера
Если скомпилировать исходники и запустить в отдельных консолях клиент и сервер, то можно обмениваться сообщениями. У этого сервера есть один значительный недостаток: он работает только с одним клиентом. Попробуем запустить в еще одной консоли новый клиент и увидим, как она зависнет. То же самое будет и с четвертым, и с пятым клиентом.
Шаг 2. Реализуем прием нескольких соединений
Чтобы принять несколько одновременных соединений, необходимо:
- функцию приема соединения
conn.Accept()заключить еще в один циклfor. - весь код, который был в цикле, вынести в отдельную функцию
process(). - запустить функцию
process()как отдельную горутину в циклеforсразу после приема соединенияconn.Accept()
В результате небольших изменений наш код примет следующий вид:
// функция process запускается как горутина
func process(conn net.Conn){
// определим, что перед выходом из функции, мы закроем соединение
defer conn.Close
for {
// Будем прослушивать все сообщения разделенные \n
message, _ := bufio.NewReader(conn).ReadString('\n')
// Распечатываем полученое сообщение
fmt.Print("Message Received:", string(message))
// Отправить новую строку обратно клиенту
conn.Write([]byte(message + "\n")){
}
}
Код в main:
// Устанавливаем прослушивание порта
ln, _ := net.Listen("tcp", ":8081")
// выполнение цикла обработки соединений
for {
// Принимаем входящее соединение
conn, _ := ln.Accept()
// запускаем функцию process(conn) как горутину
go process(conn)
}
Если мы запустим в одной консоли сервер, а в нескольких других консолях программу клиента, то можно видеть, что сервер может обрабатывать уже несколько соединений с клиентами параллельно. На самом деле он их обрабатывает асинхронно: сперва один запрос, потом другой, осуществляя переключения между горутинами.
Шаг 3. Обрабатываем ошибки соединений
Давайте попробуем отсоединить один из клиентов, убив его процесс: наш сервер зациклится. Если что-нибудь набрать в клиенте, то данные куда-то уходят, и клиент ни о чём не подозревает.
Мы забыли обработать ошибки ввода-вывода. Функция вывода в сокет Write имеет два выходных параметра: кол-во считанных байт и ошибку:
data_len, err := conn.Write(b []byte)
Если ошибка не пустая (т.е. не равна nil ), значит мы не смогли принять данные. Какая ошибка произошла, можно узнать с помощью функции err.Error()
Заменим conn.Write(b []byte) на следующий код:
_ , err := conn.Write(b []byte)
if err != nil {
fmt.Println(err.Error)
break // выходим из цикла и функции
}
Аналогичный код пропишем в клиенте. Еще в клиенте отсутствует отложенное закрытие соединения, которое срабатывает при выходе из функции defer conn.Close().
Теперь при закрытие клиента или сервера, у нас будет выдаваться сообщение:
write tcp 127.0.0.1:8081->127.0.0.1:40296: write: broken pipe
Шаг 4. Простой прототип мессенджера
К этому моменту мы допилили эхо-сервер, а теперь осталось сделать простой из него мессенджер. Пусть логика мессенджера будет следующей:
- Клиент коннектится к серверу и получает номер сообщения. Каждай клиент имеет свой уникальный номер.
- Клиент отправляет сообщение серверу с указанием номера клиента, кому адресовано это сообщение.
- Сервер принимает сообщение от клиента, декодирует его и отправляет тому клиенту, которому адресовано это сообщение.
Введем счетчик входящих соединений, а каждое новое соединение сохраним в xeштаблице, организовав таким образом пул:
conns := make( map[int] net.Conn, 1024)
Каждое соединение после conn.Accept() мы сохраним в conns, а в функцию process() будем передавать весь пул (хештаблицу) и номер текущего соединения. В функции обработки соединения process() мы можем иметь доступ ко всем активным соединениям. Не забываем увеличивать на единицу счетчик текущих соединений.
В функции process() мы принимаем не текущее соединение, а пул и номер текущего соединения. Следовательно, чтоб получить доступ к текущему соединению, мы можем его взять из пула:
conn := conn[n]
Новый код сервера:
func process(conns map[int]net.Conn, n int) {
// получаем доступ к текущему соединению
conn := conns[n]
// определим, что перед выходом из функции, мы закроем соединение
defer conn.Close()
for {
// Будем прослушивать все сообщения разделенные \n
message, _ := bufio.NewReader(conn).ReadString('\n')
// Распечатываем полученое сообщение
fmt.Print("Message Received:", string(message))
// Отправить новую строку обратно клиенту
_, err := conn.Write([]byte(strconv.Itoa(n) + "->> " + message + "\n"))
// анализируем на ошибку
if err != nil {
fmt.Print(err.Error())
break // выходим из цикла
}
}
}
func main() {
fmt.Println("Start server...")
// создаем пул соединений
conns := make(map[int]net.Conn, 1024)
i := 0
// Устанавливаем прослушивание порта
ln, _ := net.Listen("tcp", ":8081")
// объвляем пул соединений на 1024 соединения
conns := make(map[int]net.Conn, 1024)
// Запускаем цикл обработки соединений
for {
// Принимаем входящее соединение
conn, _ := ln.Accept()
// сохраняем соединение в пул
conns[i] = conn
// запускаем функцию process(conn) как горутину
go process(conns, i)
i++
}
}
При тестировании мы видим, что в каждом ответном сообщении сервер возвращает клиенту номер текущего соединения:
./client
Text to send: msg from 1
Message from server: 1->> msg from 1
Шаг 5. Реализация протокола обмена
Пример:
2 Оправляем сообщение второму клиенту
Для реализации этого протокола, необходимо сделать парсинг сообщения. Номер клиента, мы вытащим, используя fmt.Scanf(), а само сообщение с использованием слайса:
// парсинг полученного сообщения
fmt.Sscanf(message, "%d", &clientNo) // определи номер клиента
pos := strings.Index(message, " ") // нашли позицию разделителя
out_message := message[pos:] // взяли хвост сообщение после пробела
Дальше все очень просто: зная номер соединения (clientNo) клиента, мы будем отправлять ответ в нужное соединение. Сообщение было немного изменено, и теперь мы выводим, от какого клиента оно исходит:
conns[clientNo].Write([]byte(strconv.Itoa(clientNo) + "->> " + out_message + "\n"))
Шаг 6. Распараллеливание кода
Выход довольно простой: нужно чтение из сокета и чтение с консоли сделать асинхронными, т.е. сделать так, чтобы ввод из сокета и ввод из консоли не блокировали друг друга. Как говорилось выше, горутины позволяют коду выполняться асинхронно. Следовательно, должны быть запущены две разные горутины: одна должна осуществлять чтение с консоли, а вторая – чтение из сокета.
Первая горутина читает данные из сокета, и если в сокете есть данные, выводит их на печать. Чтение и вывод должны быть заключены в цикл:
// прием данных из сокета и вывод на печать
func readSock(conn net.Conn ) {
buf := make([]byte,256)
for {
readed_len, _ := conn.Read(buf)
if readed_len > 0 {
fmt.Print(string(buf))
}
}
}
Со второй горутиной все немного сложнее, поскольку она должна передать данные в основную программу. Почему нельзя сразу писать их в сокет, как это делает первая горутина? Увы, операция conn.Write() – блокирующая, и если мы так сделаем, то можем заблокировать другие операции ввода-вывода. Все блокирующие операции нужно разнести по разным асинхронным частям программы.
// ввод данных с консоли и вывод их в канал
func readConsole(ch chan string) {
for {
fmt.Print(">")
line, _ := bufio.NewReader(os.Stdin).ReadString('\n')
out :=line[:len(line)-1] // убираем символ возврата каретки
ch <- out // отправляем данные в канал
}
}
Основная программа должна запустить две асинхронных горутины: чтение с консоли и из сокета (в цикле читать канал и если в нем есть данные, то записать их в сокет). Чтобы наше консольное приложение не «съело» все ресурсы CPU, необходимо ввести некоторую задержку: time.Sleep(time.Seconds * 2)
Должно получиться примерно следующее:
func main(){
ch := make(chan string)
defer close(ch) // закрываем канал при выходе из программы
conn, _ := net.Dial("tcp", "127.0.0.1:8080")
go readConsole(ch)
go readSock(conn)
for {
val, ok := <- ch
if ok { // если есть данные, то их пишем в сокет
_, err := conn .Write([]bytes(val))
if err != nil {
fmt.Println("Write:", err.Error())
break
}
} else {
// данных в канале нет, задержка на 2 секунды
time.Sleep(time.Second * 2)
}
}
fmt.Println("Finished...")
conn.Close()
}
Шаг 7. Повышаем надежность выполнения кода
Наше приложение должно работать при любых входных данных, даже если они некорректные. Есть несколько простых правил, которые придется соблюдать при построении любых приложений:
- необходимо проверять все входящие данные и если их длинна больше принимающего буфера, то либо обрезать их, либо генерировать ошибку;
- необходимо обрабатывать все функции ввода-вывода на возможность возникновения ошибки.
Например, в коде было много сокращений и специально была опущена обработка функций conn.Accept() и net.Dial():
ls,err := conn.Accept()
if err != nil {
fmt.Println(err)
panic("accept")
}
Также был опущен код обработки объема данных с консоли:
// ввод данных с консоли
for {
fmt.Print(">")
line, _ := bufio.NewReader(os.Stdin).ReadString('\n')
if len(line) > 250 {
fmt.Println("Error: message is very lagre")
continue
}
ch <- b
}
Почти готовое и работоспособное решение можно найти в репозитории: вам остается самостоятельно дописать обработку всех ошибок ввода-вывода. Если возникнет желание сделать pull request в репозиторий, то я смогу указать в комментариях на ошибки или просто похвалить. Сделайте свой код достоянием общественности. Удачи!
Комментарии