Войти Регистрация
Конкурентность

24. Онлайн-статусы в VK Мессенджере: хаб подписок

Сложная

Go · 61 строк

Один инстанс держит около 40 тысяч открытых вкладок. Консьюмер очереди вызывает OnPresence на каждое событие, в час пик это 6 тысяч событий в секунду. serve крутится в отдельной горутине на каждую вкладку и пишет статусы в веб-сокет.

Вкладку отключают два места: ошибка записи в сокет и таймер простоя в 10 минут. Отметьте строки с ошибками.

В работе: Сервис присутствия, который показывает зелёную точку у собеседника.

Как проверяем Нажмите на строку с ошибкой и коротко напишите, что не так. Баг засчитан, если отмечена хотя бы одна его строка. Зачёт — найдено не меньше 60 % багов с учётом серьёзности (критичный весит 3, обычный 2, мелкий 1), а лишних отметок не больше, чем багов в коде. Пустые строки, скобки и комментарии лишними не считаем. Плюс 2 вопроса про поведение программы — на зачёт не влияют. Ответ один на подход: после него показываем разбор.
каналызакрытие каналадедлокпорядок событийутечка горутин

Источник: наша задача на данных VK: статусы «в сети» в мессенджере. Условие пересказано своими словами.

  1. package presence
  3. import (
  4. "sync"
  5. "time"
  6. )
  8. type Status struct {
  9. UserID int64
  10. Online bool
  11. }
  13. type Conn interface{ WriteJSON(v any) error }
  15. type Hub struct {
  16. mu sync.Mutex
  17. subs map[int64]chan Status
  18. }
  20. func (h *Hub) Subscribe(connID int64) <-chan Status {
  21. ch := make(chan Status)
  22. h.mu.Lock()
  23. h.subs[connID] = ch
  24. h.mu.Unlock()
  25. return ch
  26. }
  28. // Unsubscribe зовут ошибка записи и таймер простоя.
  29. func (h *Hub) Unsubscribe(connID int64) {
  30. h.mu.Lock()
  31. ch := h.subs[connID]
  32. delete(h.subs, connID)
  33. h.mu.Unlock()
  34. close(ch)
  35. }
  37. func (h *Hub) Publish(s Status) {
  38. h.mu.Lock()
  39. defer h.mu.Unlock()
  40. for _, ch := range h.subs {
  41. ch <- s
  42. }
  43. }
  45. // OnPresence вызывает консьюмер очереди на каждое событие.
  46. func (h *Hub) OnPresence(s Status) {
  47. go h.Publish(s)
  48. }
  50. func (h *Hub) serve(connID int64, ws Conn) {
  51. updates := h.Subscribe(connID)
  52. idle := time.AfterFunc(10*time.Minute, func() { h.Unsubscribe(connID) })
  53. defer idle.Stop()
  54. for s := range updates {
  55. idle.Reset(10 * time.Minute)
  56. if err := ws.WriteJSON(s); err != nil {
  57. h.Unsubscribe(connID)
  58. return
  59. }
  60. }
  61. }