[GoLang] gRPC를 통한 동영상 스트리밍 및 모니터링

gRPC가 얼마나 빠른지는 이 포스트에서 알아봤었고 스트리밍 방식의 종류는 이 포스트로 알아봤습니다.

이제는 동영상 전송에서의 예시를 작성해 봅시다.

 

1. Idea

동영상 파일을 단순히 전송하는 것을 넘어, 동영상 청크를 서버로 전송할 때마다 서버가 실시간으로 처리 상태(예: 처리 시간, 상태 코드 등)를 클라이언트에 전송하도록 구현해 봅니다.

이를 통해 중간 오류 감지, 타임아웃, 재시도 등의 기능을 활용하여 안정적인 동영상 스트리밍 시스템을 구성합니다.

 

gRPC 활용

• 양방향 스트리밍 활용
  클라이언트는 동영상 파일을 일정 크기의 청크로 나누어 서버로 전송하고, 서버는 각 청크를 처리한 후 즉시 처리 결과(청크 순서, 처리 상태, 소요 시간 등)를 스트리밍으로 응답합니다.
  
  
• 실시간 모니터링
  클라이언트에서는 서버로부터 전달받은 처리 결과를 실시간으로 화면(콘솔 또는 UI)에서 확인할 수 있어, 동영상 전송 중 발생할 수 있는 문제를 빠르게 인지할 수 있습니다.
  
  
• 고급 gRPC 기능 적용
  타임아웃, 컨텍스트 취소, 최대 메시지 크기 설정 등을 통해 안정성을 확보합니다.

 

2. Project Setting

프로젝트 생성

mkdir grpc_video_project
cd grpc_video_project
go mod init github.com/your_username/grpc_video_project

mkdir proto
mkdir server
mkdir client

 

 

• proto: 프로토콜 버퍼 파일과 생성된 코드를 저장
• server: gRPC 서버 코드를 작성
• client: gRPC 클라이언트 코드를 작성

 

3. Proto 

proto/ video_processing.proto 생성

syntax = "proto3";

package videopb;
option go_package = "proto/videopb;videopb";

// 양방향 스트리밍을 위한 서비스 정의
service VideoService {
  // 클라이언트가 VideoChunk 메시지를 전송하면, 서버는 ProcessingStatus 메시지를 스트리밍으로 응답
  rpc ProcessVideo(stream VideoChunk) returns (stream ProcessingStatus);
}

message VideoChunk {
  // 동영상 데이터 청크 (바이너리)
  bytes data = 1;
  // 청크 순서 번호 (전송 순서 식별용)
  int32 sequence = 2;
}

message ProcessingStatus {
  // 처리된 청크의 순서 번호
  int32 sequence = 1;
  // 처리 상태 메시지 (예: "processed")
  string status = 2;
  // 처리에 소요된 시간 (초 단위)
  float processingTime = 3;
}

proto와 Go 플러그인을 설치하고 코드를 아래 명령어로 코드를 생성합니다.

go get google.golang.org/protobuf/cmd/protoc-gen-go@latest
go get google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

protoc --go_out=. --go-grpc_out=. proto/video_processing.proto

이러면 go.mod가 다음과 유사한 형태가 됩니다.

 

만약 에러가 날 경우 아래 go.mod를 참고하세요

module github.com/your_username/grpc_video_project

go 1.24

require (
	google.golang.org/grpc v1.65.0
	google.golang.org/protobuf v1.36.6
)

require (
	golang.org/x/net v0.26.0 // indirect
	golang.org/x/sys v0.21.0 // indirect
	golang.org/x/text v0.16.0 // indirect
	google.golang.org/genproto/googleapis/rpc v0.0.0-20240604185151-ef581f913117 // indirect
)

 

4. Server

서버에서는 클라이언트가 보낸 동영상 청크를 받고 단순한 처리를 가정한 (20ms 대기) 후, 파일에 기록한 다음

처리 결과를 응답으로 보냅니다.

server/main.go 생성

package main

import (
	"fmt"
	pb "github.com/your_username/grpc_video_project/proto/videopb"
	"google.golang.org/grpc"
	"io"
	"log"
	"net"
	"os"
	"time"
)

type server struct {
	pb.UnimplementedVideoServiceServer
}

func (s *server) ProcessVideo(stream pb.VideoService_ProcessVideoServer) error {
	// 클라이언트로부터 청크를 수신하며, 각 청크마다 처리 후 상태 스트림으로 응답
	file, _ := os.Create("received_video.mp4")
	for {
		chunk, err := stream.Recv()
		if err == io.EOF {
			// 클라이언트 전송 완료
			log.Printf("모든 청크 수신 완료")
			return nil
		}
		if err != nil {
			return fmt.Errorf("청크 수신 오류: %v", err)
		}

		if _, err := file.Write(chunk.Data); err != nil {
			return fmt.Errorf("파일 기록 오류: %v", err)
		}

		// 처리 시작 시각 기록 (여기서는 처리 시뮬레이션)
		start := time.Now()
		// 예시 처리 로직: 실제 처리(예: 디코딩, 필터 적용) 대신 간단한 sleep
		time.Sleep(20 * time.Millisecond)
		processingTime := float32(time.Since(start).Seconds())

		status := &pb.ProcessingStatus{
			Sequence:       chunk.Sequence,
			Status:         "processed",
			ProcessingTime: processingTime,
		}

		// 처리 결과를 실시간으로 클라이언트에 전송
		if err := stream.Send(status); err != nil {
			return fmt.Errorf("상태 전송 오류: %v", err)
		}
		log.Printf("청크 %d 처리 완료 (%.3fs)", chunk.Sequence, processingTime)
	}
}

func main() {
	lis, err := net.Listen("tcp", ":50051")
	if err != nil {
		log.Fatalf("포트 리스닝 오류: %v", err)
	}

	grpcServer := grpc.NewServer()
	pb.RegisterVideoServiceServer(grpcServer, &server{})
	log.Println("gRPC 서버 실행 중: :50051")
	if err := grpcServer.Serve(lis); err != nil {
		log.Fatalf("서버 실행 오류: %v", err)
	}
}

 

5. Client

client/main.go

바로 서버와 동일한 포트로 연결 후, processVideo로 동영상을 스트림으로 보냅니다.

이때 파일을 읽어서 청크로 만들어 요청을 보내는 고루틴과 서버에서 주는 실시간 응답을 출력해 주는 고루틴이 있습니다.

이걸로 실시간으로 청크를 보내고 그에 대한 결과를 볼 수 있죠.

package main

import (
	"context"
	"fmt"
	"io"
	"os"
	"sync"
	"time"

	pb "github.com/your_username/grpc_video_project/proto/videopb"
	"google.golang.org/grpc"
)

// 32KB 버퍼 풀 생성 (메모리 최적화)
var bufferPool = sync.Pool{
	New: func() interface{} {
		return make([]byte, 32*1024)
	},
}

func processVideo(client pb.VideoServiceClient, filePath string) error {
	// 5분 타임아웃 설정
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()

	stream, err := client.ProcessVideo(ctx)
	if err != nil {
		return fmt.Errorf("스트림 생성 오류: %v", err)
	}

	var wg sync.WaitGroup
	wg.Add(2)

	// 1. 청크 전송 고루틴 (동영상 파일을 읽어 청크 전송)
	go func() {
		defer wg.Done()
		file, err := os.Open(filePath)
		if err != nil {
			fmt.Printf("파일 열기 오류: %v\n", err)
			return
		}
		defer file.Close()

		sequence := int32(1)
		for {
			buf := bufferPool.Get().([]byte)
			n, err := file.Read(buf)
			if err != nil {
				if err == io.EOF {
					bufferPool.Put(buf)
					break
				}
				fmt.Printf("파일 읽기 오류: %v\n", err)
				bufferPool.Put(buf)
				break
			}

			// 버퍼로 읽은 부분까지 전송
			req := &pb.VideoChunk{
				Data:     buf[:n],
				Sequence: sequence,
			}
			sequence++
			if err := stream.Send(req); err != nil {
				fmt.Printf("청크 전송 오류: %v\n", err)
				bufferPool.Put(buf)
				break
			}
			bufferPool.Put(buf)
		}
		// 전송 완료 시 스트림 종료 알림
		stream.CloseSend()
	}()

	// 2. 처리 상태 수신 고루틴 (서버의 실시간 응답을 콘솔에 출력)
	go func() {
		defer wg.Done()
		for {
			status, err := stream.Recv()
			if err == io.EOF {
				break
			}
			if err != nil {
				fmt.Printf("상태 수신 오류: %v\n", err)
				break
			}
			// 실시간 처리 결과 출력
			fmt.Printf("청크 %d: %s (%.3fs)\n", status.Sequence, status.Status, status.ProcessingTime)
		}
	}()

	wg.Wait()
	return nil
}

func main() {
	// grpc dial로 클라이언트를 생성하고
	// 포트 지정하고, withInsecure로 TLS 암호화 X
	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure(),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallRecvMsgSize(10*1024*1024),
			grpc.MaxCallSendMsgSize(10*1024*1024),
		),
	)
	if err != nil {
		fmt.Printf("연결 실패: %v\n", err)
		return
	}
	defer conn.Close()

	client := pb.NewVideoServiceClient(conn)
	if err := processVideo(client, "../video.mp4"); err != nil {
		fmt.Printf("프로세싱 오류: %v\n", err)
	}
}

 

서버와, 클라이언트의 main.go를 실행하면 프로젝트 경로의 video.mp4를 server 폴더에 received_video.mp4로 받는 걸 볼 수 있습니다.

 

서버는 다음처럼 로그가

클라이언트는 다음 처럼 로그가 쌓이는 걸 볼 수 있습니다.

 

파일 크기는 20,805kb이고 청크의 크기를 32kb로 지정했으니 딱 정확히 651번 실행되고 종료되었습니다.