마이크로서비스 통신 전략 완벽 가이드

1
2
3
4
5
6

GET    /jobs          → 작업 목록 조회
POST   /jobs          → 새 작업 생성
GET    /jobs/{id}     → 특정 작업 조회
PUT    /jobs/{id}     → 작업 전체 수정
PATCH  /jobs/{id}     → 작업 부분 수정
DELETE /jobs/{id}     → 작업 삭제

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from flask import Flask, request, jsonify
import uuid

app = Flask(__name__)
jobs = {}

@app.route('/jobs', methods=['POST'])
def create_job():
    """새로운 작업 생성"""
    data = request.get_json()
    job_id = str(uuid.uuid4())
    
    jobs[job_id] = {
        'id': job_id,
        'type': data['type'],
        'params': data['params'],
        'status': 'pending'
    }
    
    return jsonify({
        'job_id': job_id,
        'status': 'pending'
    }), 201  # 201 Created

@app.route('/jobs/<job_id>', methods=['GET'])
def get_job(job_id):
    """작업 상태 조회"""
    if job_id not in jobs:
        return jsonify({'error': 'Job not found'}), 404
    
    return jsonify(jobs[job_id]), 200

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import requests

# 작업 생성
response = requests.post(
    'http://api-server:8080/jobs',
    json={
        'type': 'data_processing',
        'params': {'input_file': 'data.csv', 'batch_size': 100}
    }
)
job = response.json()
print(f"생성된 작업 ID: {job['job_id']}")

# 결과 조회
result = requests.get(f"http://api-server:8080/jobs/{job['job_id']}")
print(result.json())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// processing.proto
syntax = "proto3";

package processing;

// 서비스 정의
service ProcessingService {
    // 단일 요청-응답
    rpc SubmitJob(JobRequest) returns (JobResponse);
    
    // 서버 스트리밍: 클라이언트 1회 요청, 서버 여러 번 응답
    rpc StreamResults(JobId) returns (stream ProcessingResult);
    
    // 양방향 스트리밍
    rpc InteractiveProcess(stream InputData) 
        returns (stream OutputData);
}

// 메시지 정의
message JobRequest {
    string job_type = 1;
    int32 batch_size = 2;
    repeated double input_data = 3;
}

message JobResponse {
    string job_id = 1;
    Status status = 2;
    
    enum Status {
        PENDING = 0;
        RUNNING = 1;
        COMPLETED = 2;
        FAILED = 3;
    }
}

message ProcessingResult {
    string chunk_id = 1;
    bytes result_data = 2;  // 바이너리 데이터
    double timestamp = 3;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import grpc
from concurrent import futures
import processing_pb2
import processing_pb2_grpc

class ProcessingServicer(processing_pb2_grpc.ProcessingServiceServicer):
    
    def SubmitJob(self, request, context):
        """단일 작업 제출"""
        job_id = create_job(request.job_type, request.batch_size)
        return processing_pb2.JobResponse(
            job_id=job_id,
            status=processing_pb2.JobResponse.PENDING
        )
    
    def StreamResults(self, request, context):
        """결과 스트리밍 - 처리가 완료될 때마다 전송"""
        for result in process_job(request.job_id):
            yield processing_pb2.ProcessingResult(
                chunk_id=result['chunk_id'],
                result_data=result['data'],
                timestamp=result['time']
            )

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    processing_pb2_grpc.add_ProcessingServiceServicer_to_server(
        ProcessingServicer(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import grpc
import processing_pb2
import processing_pb2_grpc

def run_job():
    with grpc.insecure_channel('worker-service:50051') as channel:
        stub = processing_pb2_grpc.ProcessingServiceStub(channel)
        
        # 작업 제출
        response = stub.SubmitJob(processing_pb2.JobRequest(
            job_type='batch_processing',
            batch_size=100
        ))
        print(f"작업 ID: {response.job_id}")
        
        # 결과 스트리밍 수신
        for result in stub.StreamResults(
            processing_pb2.JobId(id=response.job_id)
        ):
            print(f"Chunk {result.chunk_id}: {len(result.result_data)} bytes")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

from flask import Flask
from flask_socketio import SocketIO, emit, join_room

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")

@socketio.on('connect')
def handle_connect():
    print(f"클라이언트 연결: {request.sid}")
    emit('status', {'message': '연결 성공'})

@socketio.on('subscribe')
def handle_subscribe(data):
    """특정 채널의 실시간 업데이트 구독"""
    channel = data['channel']
    join_room(channel)
    emit('subscribed', {'channel': channel})

def broadcast_update(channel, data):
    """특정 채널에 업데이트 브로드캐스트"""
    socketio.emit('update', {
        'channel': channel,
        'data': data,
        'timestamp': time.time()
    }, room=channel)

@socketio.on('user_action')
def handle_action(data):
    """사용자 액션 수신 → 처리 결과 반환"""
    result = process_action(data['action'], data['params'])
    emit('action_result', {
        'action_id': data['action_id'],
        'result': result
    })

if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0', port=8080)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

// 웹 브라우저 클라이언트
const socket = io('http://realtime-server:8080');

socket.on('connect', () => {
    console.log('서버 연결됨');
    
    // 채널 구독
    socket.emit('subscribe', { channel: 'job-updates' });
});

socket.on('update', (data) => {
    console.log(`업데이트: ${data.channel}`);
    updateUI(data.data);
});

socket.on('action_result', (data) => {
    handleResult(data.result);
});

// 사용자 액션 전송
function sendAction(action, params) {
    socket.emit('user_action', {
        action_id: generateId(),
        action: action,
        params: params
    });
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import socketio

sio = socketio.Client()

@sio.event
def connect():
    print('서버 연결됨')
    sio.emit('subscribe', {'channel': 'job-updates'})

@sio.on('update')
def on_update(data):
    print(f"업데이트 수신: {data['channel']}")
    process_update(data['data'])

sio.connect('http://realtime-server:8080')
sio.wait()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import redis
import json
import uuid

redis_client = redis.Redis(host='redis-server', port=6379)

def submit_job(job_type, params):
    """작업을 큐에 제출"""
    job_id = str(uuid.uuid4())
    
    job_data = {
        'job_id': job_id,
        'type': job_type,
        'params': params,
        'submitted_at': time.time()
    }
    
    # 작업을 큐에 추가 (LPUSH: 왼쪽에서 삽입)
    redis_client.lpush('job_queue', json.dumps(job_data))
    
    # 작업 상태 초기화
    redis_client.hset(f'job:{job_id}', mapping={
        'status': 'queued',
        'progress': 0
    })
    
    return job_id

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

import redis
import json

redis_client = redis.Redis(host='redis-server', port=6379)

def worker_loop():
    """작업 큐에서 작업을 가져와 처리"""
    print("Worker 시작, 작업 대기 중...")
    
    while True:
        # BRPOP: 오른쪽에서 추출, 블로킹 (타임아웃 0 = 무한 대기)
        _, job_json = redis_client.brpop('job_queue', timeout=0)
        job = json.loads(job_json)
        
        job_id = job['job_id']
        print(f"작업 처리 시작: {job_id}")
        
        try:
            # 상태 업데이트
            redis_client.hset(f'job:{job_id}', 'status', 'running')
            
            # 실제 작업 수행
            result = process_job(job['type'], job['params'])
            
            # 결과 저장
            save_result(job_id, result)
            
            # 완료 상태
            redis_client.hset(f'job:{job_id}', mapping={
                'status': 'completed',
                'progress': 100
            })
            
        except Exception as e:
            redis_client.hset(f'job:{job_id}', mapping={
                'status': 'failed',
                'error': str(e)
            })

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# 우선순위별 큐 사용
QUEUES = {
    'high': 'job_queue:high',
    'normal': 'job_queue:normal', 
    'low': 'job_queue:low'
}

def submit_job(job_data, priority='normal'):
    redis_client.lpush(QUEUES[priority], json.dumps(job_data))

def worker_with_priority():
    """우선순위 순서로 작업 처리"""
    while True:
        # 높은 우선순위부터 확인
        result = redis_client.brpop(
            [QUEUES['high'], QUEUES['normal'], QUEUES['low']],
            timeout=5
        )
        if result:
            queue_name, job_json = result
            process_job(json.loads(job_json))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import pika

# 연결 설정
connection = pika.BlockingConnection(
    pika.ConnectionParameters('rabbitmq-server')
)
channel = connection.channel()

# 교환기와 큐 선언
channel.exchange_declare(exchange='jobs', exchange_type='topic')
channel.queue_declare(queue='image_processing', durable=True)
channel.queue_bind(
    exchange='jobs',
    queue='image_processing',
    routing_key='job.image.*'
)

# 메시지 발행
channel.basic_publish(
    exchange='jobs',
    routing_key='job.image.resize',
    body=json.dumps(job_data),
    properties=pika.BasicProperties(delivery_mode=2)  # 메시지 영속성
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# Realtime Server: gRPC 스트리밍 → WebSocket 브릿지
import grpc
import processing_pb2
import processing_pb2_grpc
from flask_socketio import SocketIO, emit

@socketio.on('subscribe_job')
def handle_subscribe(data):
    """gRPC 스트리밍을 받아 WebSocket으로 중계"""
    job_id = data['job_id']

    # 내부 gRPC 서비스에 스트리밍 요청
    channel = grpc.insecure_channel('processing-service:50051')
    stub = processing_pb2_grpc.ProcessingServiceStub(channel)

    for result in stub.StreamResults(
        processing_pb2.JobId(id=job_id)
    ):
        # gRPC 결과를 WebSocket으로 실시간 전달
        emit('job_progress', {
            'job_id': job_id,
            'chunk_id': result.chunk_id,
            'data_size': len(result.result_data),
            'timestamp': result.timestamp
        })