Go 채팅 응용 프로그램에서 Redis 게시/구독을 사용하는 방법 (3 섹션)

이 강좌의 세 번째 부분에서 우리는 Redis 발표/구독을 기존 채팅 응용 프로그램에 추가할 것이다.Redis 게시/구독을 사용하면 여러 인스턴스를 동시에 실행하여 애플리케이션을 확장할 수 있습니다.

전제 조건

계속하려면 part 1 및 part 2 또는 here 에서 소스 코드를 가져와야 합니다.

Redis Pub/Sub란 무엇입니까?

Reds 게시/구독은 게시 - 구독 모드의 Redis 구현입니다.이것은 이른바'메시지 전달 모델'이다. 메시지의 발송자(발표자)는 수신자(구독자)에게 직접 메시지를 보내지 않고'통로'에서 메시지를 발표한다.구독자는 특정 채널에 구독을 선택하고 발표된 메시지를 받을 것입니다.
우리가 같은 응용 프로그램의 여러 개의 실례를 실행할 때, 우리는 이러한 발표/구독 채널을 이용하여 같은 실례에 연결된 클라이언트뿐만 아니라, 모든 실례에 연결된 모든 클라이언트에게도 통지할 수 있다.

게시/가입 다이어그램.
우리의 응용 프로그램에 대해 모든 채팅 메시지는 채팅방을 통해 발송되기 때문에 우리는 이 채팅방을 이용하여 자신의 채널에서 발표하고 구독할 수 있다.따라서 모든 방에 술집/하위 채널이 있습니다. (위 그림의 방 채널 참조)
우리는 명단에 있는 모든 온라인 사용자가 서버마다, 그리고 개인 채팅을 시작할 수 있기를 희망한다. 예를 들어.이를 위해 WsServer에서 게시하고 구독할 수 있는 공통 채널을 사용합니다.네, 인코딩을 시작합시다!

1단계: 지구층 추가

게시/구독은 놓친 메시지를 재생하지 않기 때문에 우리는 어떤 지속성이 필요하다.만약 우리가 서비스가 실행된 후에 응용 프로그램을 확장한다면, 새로운 실례는 모든 기존 데이터 (방과 사용자) 를 얻는 방법이 필요하다.
이를 위해 우리는 데이터베이스를 추가할 것입니다. 이 글에서 우리는 간단함을 유지하고 SQLite 데이터베이스를 사용할 것입니다.용례에 따라 다른 데이터베이스 엔진을 사용해야 합니다.교환이 편리하도록 우리는 저장소 모드를 사용할 것이다.
설치에 필요한 소프트웨어 패키지:

go get github.com/mattn/go-sqlite3

// config/database.go
package config

import (
    "database/sql"
    "log"

    _ "github.com/mattn/go-sqlite3"
)

func InitDB() *sql.DB {
    db, err := sql.Open("sqlite3", "./chatdb.db")
    if err != nil {
        log.Fatal(err)
    }

    sqlStmt := `    
    CREATE TABLE IF NOT EXISTS room (
        id VARCHAR(255) NOT NULL PRIMARY KEY,
        name VARCHAR(255) NOT NULL,
        private TINYINT NULL
    );
    `
    _, err = db.Exec(sqlStmt)
    if err != nil {
        log.Fatal("%q: %s\n", err, sqlStmt)
    }

    sqlStmt = ` 
    CREATE TABLE IF NOT EXISTS user (
        id VARCHAR(255) NOT NULL PRIMARY KEY,
        name VARCHAR(255) NOT NULL
    );
    `
    _, err = db.Exec(sqlStmt)
    if err != nil {
        log.Fatal("%q: %s\n", err, sqlStmt)
    }

    return db
}


// main.go
..
import (
    ...
    "github.com/jeroendk/chatApplication/config"
    "github.com/jeroendk/chatApplication/repository"
)

func main() {
    ...
    db := config.InitDB()
    defer db.Close()
}

위의 코드는 Go 응용 프로그램을 시작할 때 데이터베이스를 초기화합니다.

룸 저장소

다음은 두 개의 저장소 파일을 추가합니다. 첫 번째는roomRepository입니다.모든 가방에서 방 모형을 사용할 수 있도록 모형 가방에 인터페이스를 만들 것입니다.우리는 또한 룸 Repository에 인터페이스를 추가해서 교체를 더욱 쉽게 할 수 있다.

// models/room.go
package models

type Room interface {
    GetId() string
    GetName() string
    GetPrivate() bool
}

type RoomRepository interface {
    AddRoom(room Room)
    FindRoomByName(name string) Room
}

// repository/roomRepository.go

package repository

import (
    "database/sql"

    "github.com/jeroendk/chatApplication/models"
)

type Room struct {
    Id string
    Name string
    Private bool
}

func (room *Room) GetId() string {
    return room.Id
}

func (room *Room) GetName() string {
    return room.Name
}

func (room *Room) GetPrivate() bool {
    return room.Private
}

type RoomRepository struct {
    Db *sql.DB
}

func (repo *RoomRepository) AddRoom(room models.Room) {
    stmt, err := repo.Db.Prepare("INSERT INTO room(id, name, private) values(?,?,?)")
    checkErr(err)

    _, err = stmt.Exec(room.GetId(), room.GetName(), room.GetPrivate())
    checkErr(err)
}

func (repo *RoomRepository) FindRoomByName(name string) models.Room {

    row := repo.Db.QueryRow("SELECT id, name, private FROM room where name = ? LIMIT 1", name)

    var room Room

    if err := row.Scan(&room.Id, &room.Name, &room.Private); err != nil {
        if err == sql.ErrNoRows {
            return nil
        }
        panic(err)
    }

    return &room

}

func checkErr(err error) {
    if err != nil {
        panic(err)
    }
}

저장소 파일에는 두 가지 방법이 있습니다. 하나는 새 방을 추가하는 것이고, 다른 하나는 주어진 이름에 따라 방을 찾는 것입니다.

사용자 저장소

우리는 사용자를 위해 같은 일을 하고 인터페이스를 추가하며 저장소를 만들 것이다.

// models/user.go
package models

type User interface {
    GetId() string
    GetName() string
}

type UserRepository interface {
    AddUser(user User)
    RemoveUser(user User)
    FindUserById(ID string) User
    GetAllUsers() []User
}


package repository

import (
    "database/sql"
    "log"

    "github.com/jeroendk/chatApplication/models"
)

type User struct {
    Id string `json:"id"`
    Name string `json:"name"`
}

func (user *User) GetId() string {
    return user.Id
}

func (user *User) GetName() string {
    return user.Name
}

type UserRepository struct {
    Db *sql.DB
}

func (repo *UserRepository) AddUser(user models.User) {
    stmt, err := repo.Db.Prepare("INSERT INTO user(id, name) values(?,?)")
    checkErr(err)

    _, err = stmt.Exec(user.GetId(), user.GetName())
    checkErr(err)
}

func (repo *UserRepository) RemoveUser(user models.User) {
    stmt, err := repo.Db.Prepare("DELETE FROM user WHERE id = ?")
    checkErr(err)

    _, err = stmt.Exec(user.GetId())
    checkErr(err)
}

func (repo *UserRepository) FindUserById(ID string) models.User {

    row := repo.Db.QueryRow("SELECT id, name FROM user where id = ? LIMIT 1", ID)

    var user User

    if err := row.Scan(&user.Id, &user.Name); err != nil {
        if err == sql.ErrNoRows {
            return nil
        }
        panic(err)
    }

    return &user

}

func (repo *UserRepository) GetAllUsers() []models.User {

    rows, err := repo.Db.Query("SELECT id, name FROM user")

    if err != nil {
        log.Fatal(err)
    }
    var users []models.User
    defer rows.Close()
    for rows.Next() {
        var user User
        rows.Scan(&user.Id, &user.Name)
        users = append(users, &user)
    }

    return users
}

사용자 저장소에는 다음과 같은 네 가지 방법이 있습니다.

AddUser, 새 사용자를 데이터베이스에 추가합니다.

RemoveUser, 데이터베이스에서 사용자를 제거합니다.

FindUserById, 지정된 ID를 통해 사용자를 찾습니다.

GetAllUsers, 데이터베이스에서 모든 사용자를 검색합니다.

인터페이스를 사용하기 위해 기존 코드 업데이트

계속하기 전에, 우리는 우선 새로운 인터페이스에 부합하기 위해 기존의 코드를 업데이트해야 한다.

소식

// message.go
import (
    ...
    "github.com/jeroendk/chatApplication/models"
)

... 

type Message struct {
    Action string `json:"action"`
    Message string `json:"message"`
    Target *Room `json:"target"`
    Sender models.User `json:"sender"` // Use model.User interface
}

...

// UnmarshalJSON custom unmarshel to create a Client instance for Sender 
func (message *Message) UnmarshalJSON(data []byte) error {
    type Alias Message
    msg := &struct {
        Sender Client `json:"sender"`
        *Alias
    }{
        Alias: (*Alias)(message),
    }
    if err := json.Unmarshal(data, &msg); err != nil {
        return err
    }
    message.Sender = &msg.Sender
    return nil
}

고객

// client.go
import (
    ...
    "github.com/jeroendk/chatApplication/models"
)

// Change the type sender from Client to the User interface.
func (client *Client) joinRoom(roomName string, sender models.User) {
  ...
}

func (client *Client) notifyRoomJoined(room *Room, sender models.User) {
  ...
}

// Add the GetId method to make Client compatible with model.User interface
func (client *Client) GetId() string {
    return client.ID.String()
}

방.

// room.go

// Add the GetPrivate method to make Room compatible with model.Room interface
func (room *Room) GetPrivate() bool {
    return room.Private
}

2단계: 저장소 사용

현재 채팅 서버는 사용자와 채팅방을 추적하는 것을 맡고 있다.그것은 이 실체들을 지도(클라이언트와 방)에 놓아서 실현한다.우리는 계속 이렇게 할 것이다. 그러나 가장 중요한 것은 이 두 실체를 데이터베이스에 쓰는 것이다.
초보자의 경우 이 두 저장소를 구조에 속성으로 추가하고 New Websocket Server 방법에 설정합니다.모든 사용자를 추적하기 위해 새 속성'users'를 추가했습니다.clients 속성은 활성 WebSocket 연결이 있는 실제 클라이언트에 사용됩니다. (이것은 발표/구독 논리를 위한 것입니다.)

// chatServer.go
import (    
    "github.com/jeroendk/chatApplication/models"
)

type WsServer struct {
    ...
    users []models.User
    roomRepository models.RoomRepository
    userRepository models.UserRepository
}

func NewWebsocketServer(roomRepository models.RoomRepository, userRepository models.UserRepository) *WsServer {
    wsServer := &WsServer{
        clients: make(map[*Client]bool),
        register: make(chan *Client),
        unregister: make(chan *Client),
        rooms: make(map[*Room]bool),
        roomRepository: roomRepository,
        userRepository: userRepository,
    }

    // Add users from database to server
    wsServer.users = userRepository.GetAllUsers()

    return wsServer
}

WsServer의 새 인스턴스를 만들 때 모든 사용자가 데이터베이스에서 로드됩니다.
다음 단계는 main에서 te 호출을 New Websocket Server로 변경하는 것입니다.이 두 개의 저장소를 포함하러 가다

// main.go
...
wsServer := NewWebsocketServer(&repository.RoomRepository{Db: db}, &repository.UserRepository{Db: db})

룸 저장소 사용

이제 저장소에 접근할 수 있습니다. chatServer 방법에서 사용할 수 있습니다.우선, 우리는 모든 기존 방법을 업데이트하여userRepository를 사용할 것이다.다음은 수정된 방법으로 새 코드에 주석으로 표시합니다.

// chatServer.go

func (server *WsServer) registerClient(client *Client) {
    // NEW: Add user to the repo
    server.userRepository.AddUser(client)    

    // Existing actions
    server.notifyClientJoined(client)
    server.listOnlineClients(client)
    server.clients[client] = true

    // NEW: Add user to the user slice
    server.users = append(server.users, message.Sender)
}

func (server *WsServer) unregisterClient(client *Client) {
    if _, ok := server.clients[client]; ok {
        delete(server.clients, client)
        server.notifyClientLeft(client)

        // NEW: Remove user from slice
        for i, user := range server.users {
          if user.GetId() == message.Sender.GetId() {
            server.users[i] = server.users[len(server.users)-1]
            server.users = server.users[:len(server.users)-1]
          }
        }

        // NEW: Remove user from repo
        server.userRepository.RemoveUser(client)
    }
}

func (server *WsServer) listOnlineClients(client *Client) {
    // NEW: Use the users slice instead of the client map
    for _, user := range server.users {
      message := &Message{
        Action: UserJoinedAction,
        Sender: user,
      }
      client.send <- message.encode()
    }
}

상기 내용을 추가하면 모든 온라인 사용자는 데이터베이스에 저장해야 한다.사용자가 연결을 끊으면 데이터베이스에서 삭제됩니다.

사용자 저장소 사용

다음은 방.서버를 시작할 때, 우리는 모든 방을 필요로 하지 않는다.따라서 로컬 지도에서 찾을 수 없을 때만 저장소에서 찾을 수 있습니다.

// chatServer.go

func (server *WsServer) findRoomByName(name string) *Room {
    var foundRoom *Room
    for room := range server.rooms {
        if room.GetName() == name {
            foundRoom = room
            break
        }
    }

    // NEW: if there is no room, try to create it from the repo
    if foundRoom == nil {
        // Try to run the room from the repository, if it is found.
        foundRoom = server.runRoomFromRepository(name)
    }

    return foundRoom
}

// NEW: Try to find a room in the repo, if found Run it.
func (server *WsServer) runRoomFromRepository(name string) *Room {
    var room *Room
    dbRoom := server.roomRepository.FindRoomByName(name)
    if dbRoom != nil {
        room = NewRoom(dbRoom.GetName(), dbRoom.GetPrivate())
        room.ID, _ = uuid.Parse(dbRoom.GetId())

        go room.RunRoom()
        server.rooms[room] = true
    }

    return room
}

func (server *WsServer) createRoom(name string, private bool) *Room {
    room := NewRoom(name, private)
    // NEW: Add room to repo
    server.roomRepository.AddRoom(room)

    go room.RunRoom()
    server.rooms[room] = true

    return room
}

그렇습니다. 다음 단계에서 우리는 최종적으로 발표/구독 통합을 추가할 것입니다.

3단계: Redis 게시 / 구독

이제 모든 것이 준비되었습니다. Redis 게시/구독 채널을 추가하고 구독할 수 있습니다.
먼저 Redis 패키지를 설치합니다.

go mod init
go get github.com/go-redis/redis/v8

그런 다음 사용할 Redis 컨테이너가 있는지 확인합니다.docker와 docker compose를 사용하여 다음을 만들 수 있습니다.

# docker-compose.yml
version: '3.5'

services:
  redis:
    image: "redis:alpine"
    ports:
      - "6364:6379"

그리고 Docker compose up부터 시작해요.
Redis 컨테이너가 시작되고 실행됨에 따라 응용 프로그램에 연결을 만듭니다.이를 위해, 우리는 redis라는 새 파일을 만들 것입니다.데이터베이스에 연결된 config 폴더에 넣읍시다.

// config/redis.go

package config

import "github.com/go-redis/redis/v8"

var Redis *redis.Client

func CreateRedisClient() {
    opt, err := redis.ParseURL("redis://localhost:6364/0")
    if err != nil {
        panic(err)
    }

    redis := redis.NewClient(opt)
    Redis = redis
}

그리고 메인 서버에서 연결을 초기화합니다.가다

// main.go

func main() {
    ...
    config.CreateRedisClient()
    ...
}

우리는 발표/구독 채널을 통해 총 4개의 다른 메시지를 보내기를 희망한다.

채팅 메시지

사용자 가입 알림

사용자 이탈 알림

개인 채팅 초대

채팅 메시지

방 안에서 채팅 메시지를 보내는 것은 우리 방의 일이다.가다실제로 이런 논리에서는 게시/구독 채널을 통합하기 쉽다.
우선, 우리는 채널에 채널을 발표하고 구독하는 두 가지 새로운 방법을 추가할 것이다.

// room.go
package main
import (
    "fmt"
    "log"
    "github.com/jeroendk/chatApplication/config"
    "github.com/google/uuid"
    "context"
)

var ctx = context.Background()

...
func (room *Room) publishRoomMessage(message []byte) {
    err := config.Redis.Publish(ctx, room.GetName(), message).Err()

    if err != nil {
        log.Println(err)
    }
}

func (room *Room) subscribeToRoomMessages() {
    pubsub := config.Redis.Subscribe(ctx, room.GetName())

    ch := pubsub.Channel()

    for msg := range ch {
        room.broadcastToClientsInRoom([]byte(msg.Payload))
    }
}

그리고 Broadcast To Clients Room에 대한 기존 호출을 변경할 것입니다. 대신 새로운 발표 방법을 사용할 것입니다.또한 채팅방을 시작할 때 게시/구독 목록을 시작합니다.

// room.go 
func (room *Room) RunRoom() {
    // subscribe to pub/sub messages inside a new goroutine
    go room.subscribeToRoomMessages()

    for {
        select {
        ...
        case message := <-room.broadcast:
            room.publishRoomMessage(message.encode())
        }
    }
}

func (room *Room) notifyClientJoined(client *Client) {
    ...
    room.publishRoomMessage(message.encode())
}

사용자가 왼쪽 가입 (&L)

다음은 사용자가 채팅 서버에 가입하고 이 이벤트들을 구독할 때 발표합니다.가다

// chatServer.go
package main

import (
    "encoding/json"
    "log"

    "github.com/google/uuid"
    "github.com/jeroendk/chatApplication/config"
    "github.com/jeroendk/chatApplication/models"
)

const PubSubGeneralChannel = "general"

// Publish userJoined message in pub/sub
func (server *WsServer) publishClientJoined(client *Client) {

    message := &Message{
        Action: UserJoinedAction,
        Sender: client,
    }

    if err := config.Redis.Publish(ctx, PubSubGeneralChannel, message.encode()).Err(); err != nil {
        log.Println(err)
    }
}

// Publish userleft message in pub/sub
func (server *WsServer) publishClientLeft(client *Client) {

    message := &Message{
        Action: UserLeftAction,
        Sender: client,
    }

    if err := config.Redis.Publish(ctx, PubSubGeneralChannel, message.encode()).Err(); err != nil {
        log.Println(err)
    }
}

// Listen to pub/sub general channels
func (server *WsServer) listenPubSubChannel() {

    pubsub := config.Redis.Subscribe(ctx, PubSubGeneralChannel)
    ch := pubsub.Channel()
    for msg := range ch {

        var message Message
        if err := json.Unmarshal([]byte(msg.Payload), &message); err != nil {
            log.Printf("Error on unmarshal JSON message %s", err)
            return
        }

        switch message.Action {
        case UserJoinedAction:
            server.handleUserJoined(message)
        case UserLeftAction:
            server.handleUserLeft(message)      
        }
    }
}

func (server *WsServer) handleUserJoined(message Message) {
    // Add the user to the slice
    server.users = append(server.users, message.Sender)
    server.broadcastToClients(message.encode())
}

func (server *WsServer) handleUserLeft(message Message) {
    // Remove the user from the slice
    for i, user := range server.users {
        if user.GetId() == message.Sender.GetId() {
            server.users[i] = server.users[len(server.users)-1]
            server.users = server.users[:len(server.users)-1]
        }
    }
    server.broadcastToClients(message.encode())
}

publishClientJoined와 publishClientLeft는 notifyClientJoined와 notifyClientLeft를 대체합니다.
그런 다음 채널을 다시 시작하여 게시 방법을 올바르게 사용했는지 확인합니다.

// chatServer.go
func (server *WsServer) Run() {
    go server.listenPubSubChannel()
    ...
}

func (server *WsServer) registerClient(client *Client) {
    // Add user to the repo
    server.userRepository.AddUser(client)

    // Publish user in PubSub
    server.publishClientJoined(client)

    server.listOnlineClients(client)
    server.clients[client] = true
}

func (server *WsServer) unregisterClient(client *Client) {
    if _, ok := server.clients[client]; ok {
        delete(server.clients, client)

        // Remove user from repo
        server.userRepository.RemoveUser(client)

        // Publish user left in PubSub
        server.publishClientLeft(client)
    }
}

개인 채팅

거의 완성되지 않았는데, 마지막 난제는 우리 사용자들이 서로 다른 서버에 연결할 때 서로 개인 채팅을 시작하는 것이다.
클라이언트의 논리를 변경하여 시작합니다.가다

// client.go

import (    
    ...
    "github.com/jeroendk/chatApplication/config"
    ...
)

func (client *Client) handleJoinRoomPrivateMessage(message Message) {
    // instead of searching for a client, search for User by the given ID.
    target := client.wsServer.findUserByID(message.Message)
    if target == nil {
        return
    }

    // create unique room name combined to the two IDs
    roomName := message.Message + client.ID.String()

    // Join room
    joinedRoom := client.joinRoom(roomName, target)

    // Instead of instantaneously joining the target client. 
    // Let the target client join with a invite request over pub/sub
    if joinedRoom != nil {
        client.inviteTargetUser(target, joinedRoom)
    }
}

// JoinRoom now returns a room or nil
func (client *Client) joinRoom(roomName string, sender models.User) *Room {

    room := client.wsServer.findRoomByName(roomName)
    if room == nil {
        room = client.wsServer.createRoom(roomName, sender != nil)
    }

    // Don't allow to join private rooms through public room message
    if sender == nil && room.Private {
        return nil
    }

    if !client.isInRoom(room) {
        client.rooms[room] = true
        room.register <- client
        client.notifyRoomJoined(room, sender)
    }
    return room
}

// Send out invite message over pub/sub in the general channel.
func (client *Client) inviteTargetUser(target models.User, room *Room) {
    inviteMessage := &Message{
        Action: JoinRoomPrivateAction,
        Message: target.GetId(),
        Target: room,
        Sender: client,
    }

    if err := config.Redis.Publish(ctx, PubSubGeneralChannel, inviteMessage.encode()).Err(); err != nil {
        log.Println(err)
    }
}

따라서 우리 고객들은 다시 한 번 개인 채팅을 시작할 수 있다.우리가 지금 해야 할 일은 목표 고객도 가입할 수 있도록 확보하는 것이다.
다음 코드를 채팅 서버에 추가합니다.가다첫 번째 부분은 교환기에 개인 채팅 요청을 처리하는 데 사용되는 추가 사례를 추가했다.

// chatServer.go
func (server *WsServer) listenPubSubChannel() {
    ...

        switch message.Action {
        ...
        case JoinRoomPrivateAction:
            server.handleUserJoinPrivate(message)
        }
}

func (server *WsServer) handleUserJoinPrivate(message Message) {
    // Find client for given user, if found add the user to the room.
    targetClient := server.findClientByID(message.Message)
    if targetClient != nil {
        targetClient.joinRoom(message.Target.GetName(), message.Sender)
    }
}

// Add the findUserByID method used by client.go
func (server *WsServer) findUserByID(ID string) models.User {
    var foundUser models.User
    for _, client := range server.users {
        if client.GetId() == ID {
            foundUser = client
            break
        }
    }

    return foundUser
}

결과

새 설정을 테스트하려면 서로 다른 포트에서 프로그램을 시작할 수 있는 여러 개의 실례가 있습니다.Javascript WebSocket이 실제 서버에 연결되어 있는지 확인합니다.다음과 같이 연결 문자열을 변경할 수 있습니다.

serverUrl: "ws://" + location.host + "/ws",

다음:

go run ./ --addr=:8080
go run ./ --addr=:8090

완성!Go에서 게시/구독 채팅 응용 프로그램을 완료했습니다.이 시리즈의 마지막 부분을 계속 지켜봐 주십시오.그곳에서 우리는 사용자를 로그인시킨 후에야 채팅에 참여할 수 있다.
사용자가 어느 정도 잠시 중단된 후에 자동으로 다시 연결하고 싶다면 this 를 보십시오.
만약 당신에게 어떤 건의나 문제가 있다면, 언제든지 논평을 발표하세요!
이 섹션의 최종 소스 코드는 다음과 같습니다.
https://github.com/jeroendk/go-vuejs-chat/tree/v3.0