Semaphore

 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

import threading
import time
from contextlib import contextmanager

# 동시 5개까지 허용
sem = threading.BoundedSemaphore(value=5)

@contextmanager
def acquire_slot(timeout=None):
    # tryAcquire 유사: timeout 없으면 블로킹
    ok = sem.acquire(timeout=timeout) if timeout else sem.acquire()
    try:
        if not ok:
            raise TimeoutError("No permit within timeout")
        yield
    finally:
        # BoundedSemaphore는 과다 release 시 예외로 잡아줌
        sem.release()

def handle_request(i):
    try:
        with acquire_slot(timeout=0.2):
            # 임계 구역: 실제 작업
            time.sleep(0.05)
            print(f"handled {i}")
    except TimeoutError:
        print(f"throttled {i}")

threads = [threading.Thread(target=handle_request, args=(i,)) for i in range(20)]
[t.start() for t in threads]
[t.join() for t in threads]

 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

// go get golang.org/x/sync/semaphore
package main

import (
	"context"
	"fmt"
	"time"
	"golang.org/x/sync/semaphore"
)

func main() {
	// 총 허용 가중치 10
	sem := semaphore.NewWeighted(10)
	ctx := context.Background()

	// 비용이 다른 작업들
	jobs := []int64{3, 2, 5, 4, 6}

	for i, w := range jobs {
		go func(id int, weight int64) {
			// 타임아웃으로 응답성 보장
			cctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
			defer cancel()

			if err := sem.Acquire(cctx, weight); err != nil {
				fmt.Println("throttled", id)
				return
			}
			// 임계 구역
			time.Sleep(50 * time.Millisecond)
			sem.Release(weight)
			fmt.Println("done", id)
		}(i, w)
	}
	time.Sleep(1 * time.Second)
}

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

using System;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;

class ThrottledClient {
    private readonly SemaphoreSlim _sem = new SemaphoreSlim(8); // 동시 8
    private readonly HttpClient _http = new HttpClient();

    public async Task<string> GetAsync(string url, CancellationToken ct) {
        // 타임아웃/취소 가능
        using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct);
        linked.CancelAfter(TimeSpan.FromMilliseconds(300));
        await _sem.WaitAsync(linked.Token);

        try {
            return await _http.GetStringAsync(url, linked.Token);
        }
        finally {
            _sem.Release();
        }
    }
}

 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

class Semaphore {
  constructor(permits) {
    this.permits = permits;
    this.queue = [];
  }
  async acquire(timeoutMs) {
    if (this.permits > 0) { this.permits--; return; }
    let timer; 
    return new Promise((resolve, reject) => {
      const ticket = () => { clearTimeout(timer); resolve(); };
      this.queue.push(ticket);
      if (timeoutMs != null) {
        timer = setTimeout(() => {
          const idx = this.queue.indexOf(ticket);
          if (idx >= 0) this.queue.splice(idx, 1);
          reject(new Error("timeout"));
        }, timeoutMs);
      }
    });
  }
  release() {
    if (this.queue.length > 0) {
      const next = this.queue.shift();
      next(); // 깨우기
    } else {
      this.permits++;
    }
  }
}

// 사용 예: 동시 4개 파일 처리
const sem = new Semaphore(4);
async function work(id) {
  try {
    await sem.acquire(200);
    await new Promise(r => setTimeout(r, 50)); // 임계 구역
    console.log("done", id);
  } catch { console.log("throttled", id); }
  finally { sem.release(); }
}
for (let i=0;i<12;i++) work(i);

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

import threading

# 최대 5개 커넥션 제한 세마포어 생성
db_semaphore = threading.Semaphore(5)

def db_access():
    # 세마포어 wait(P) 역할
    db_semaphore.acquire()
    try:
        # 실제 DB 작업 수행
        print("DB 작업 수행 중")
        # … (DB 로직)
    finally:
        # 세마포어 signal(V) 역할 (자원 반환)
        db_semaphore.release()

# 여러 스레드 시뮬레이션
threads = [threading.Thread(target=db_access) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

 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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

import asyncio
import aiohttp
from asyncio import Semaphore
from typing import List, Dict

class WebCrawler:
    def __init__(self, max_concurrent_requests: int = 5):
        # 세마포어로 동시 요청 수 제한
        self.semaphore = Semaphore(max_concurrent_requests)
        self.session = None
        self.results = []
    
    async def fetch_url(self, url: str) -> Dict:
        """세마포어를 사용한 안전한 URL 페치"""
        # 세마포어 획득 - 동시 실행 수 제한
        async with self.semaphore:
            try:
                print(f"요청 시작: {url}")
                async with self.session.get(url, timeout=10) as response:
                    content = await response.text()
                    print(f"요청 완료: {url} (상태: {response.status})")
                    return {
                        'url': url,
                        'status': response.status,
                        'content_length': len(content),
                        'success': True
                    }
            except Exception as e:
                print(f"요청 실패: {url} - {str(e)}")
                return {
                    'url': url,
                    'error': str(e),
                    'success': False
                }
            # 세마포어는 async with 블록 종료 시 자동 release
    
    async def crawl_urls(self, urls: List[str]) -> List[Dict]:
        """여러 URL을 동시에 크롤링 (세마포어로 제한)"""
        async with aiohttp.ClientSession() as session:
            self.session = session
            
            # 모든 URL에 대해 비동기 작업 생성
            tasks = [self.fetch_url(url) for url in urls]
            
            # 세마포어가 동시 실행 수를 자동 제어
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            return results

# 사용 예시
async def main():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2', 
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/3',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1',
    ]
    
    crawler = WebCrawler(max_concurrent_requests=3)  # 최대 3개 동시 요청
    
    print("크롤링 시작...")
    start_time = asyncio.get_event_loop().time()
    
    results = await crawler.crawl_urls(urls)
    
    end_time = asyncio.get_event_loop().time()
    print(f"\n총 소요 시간: {end_time - start_time:.2f}초")
    print(f"성공한 요청: {sum(1 for r in results if r.get('success', False))}개")

# 실행
if __name__ == "__main__":
    asyncio.run(main())

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

const { Semaphore } = require('await-semaphore');
const semaphore = new Semaphore(3); // 3 concurrent requests max

async function callApi() {
    const [value, release] = await semaphore.acquire();
    try {
        // 실제 API 호출
        console.log("API 호출 실행");
        // … (API 로직)
    } finally {
        release(); // 세마포어 반환
    }
}
// 다중 API 요청 시뮬레이션
for (let i = 0; i < 10; i++) {
    callApi();
}

 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

import threading, time, random

BUFFER_SIZE = 5
buffer = []

# 세마포어 생성
empty = threading.Semaphore(BUFFER_SIZE)  # 남은 공간
full = threading.Semaphore(0)             # 채워진 데이터
mutex = threading.Lock()                   # 버퍼 접근 보호

def producer(pid):
    while True:
        item = random.randint(1, 100)
        empty.acquire()  # 빈 공간 확인
        with mutex:      # 버퍼 접근 동기화
            buffer.append(item)
            print(f"[생산자 {pid}] 데이터 {item} 생산. 버퍼 상태: {buffer}")
        full.release()   # 데이터 증가 알림
        time.sleep(random.uniform(0.5, 1.5))

def consumer(cid):
    while True:
        full.acquire()   # 데이터 존재 확인
        with mutex:
            item = buffer.pop(0)
            print(f"[소비자 {cid}] 데이터 {item} 소비. 버퍼 상태: {buffer}")
        empty.release()  # 빈 공간 증가 알림
        time.sleep(random.uniform(0.5, 2.0))

# 스레드 실행
for i in range(2):
    threading.Thread(target=producer, args=(i,)).start()
for i in range(3):
    threading.Thread(target=consumer, args=(i,)).start()

 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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

import redis
import time
import random

class DistributedSemaphore:
    def __init__(self, redis_client, name, limit):
        self.redis = redis_client
        self.name = f"semaphore:{name}"
        self.limit = limit

    def acquire(self):
        """세마포어 값이 제한보다 작을 때만 진입"""
        while True:
            self.redis.watch(self.name)
            current = int(self.redis.get(self.name) or 0)
            if current < self.limit:
                pipe = self.redis.pipeline()
                pipe.multi()
                pipe.incr(self.name)
                pipe.execute()
                return True
            self.redis.unwatch()
            time.sleep(0.05)

    def release(self):
        """자원 해제"""
        self.redis.decr(self.name)

# Redis 연결
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

BUFFER_KEY = "shared_buffer"
BUFFER_SIZE = 5

empty_sem = DistributedSemaphore(r, "empty", BUFFER_SIZE)
full_sem = DistributedSemaphore(r, "full", 0)

def producer(pid):
    while True:
        item = random.randint(1, 100)
        empty_sem.acquire()  # 빈 공간 확보
        r.lpush(BUFFER_KEY, item)
        print(f"[Producer {pid}] 생산 → {item}")
        full_sem.release()  # 데이터 개수 증가
        time.sleep(random.uniform(0.5, 1.5))

def consumer(cid):
    while True:
        full_sem.acquire()  # 데이터 확보
        item = r.rpop(BUFFER_KEY)
        if item:
            print(f"[Consumer {cid}] 소비 ← {item}")
            empty_sem.release()
        time.sleep(random.uniform(0.5, 2.0))

# 예시 실행 (단일 프로세스 내 여러 스레드, 분산 시에는 각기 다른 서버에서 실행)
import threading
threading.Thread(target=producer, args=(1,)).start()
threading.Thread(target=consumer, args=(1,)).start()

 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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

from kafka import KafkaConsumer  # Kafka 파이썬 클라이언트
import redis
import time
import json

class DistributedSemaphore:
    def __init__(self, redis_client, name, limit):
        self.redis = redis_client
        self.name = f"semaphore:{name}"
        self.limit = limit

    def acquire(self):
        """해당 세마포어 자원을 획득 (제한 이하면 허용)"""
        while True:
            self.redis.watch(self.name)
            current = int(self.redis.get(self.name) or 0)
            if current < self.limit:
                pipe = self.redis.pipeline()
                pipe.multi()
                pipe.incr(self.name)
                pipe.execute()
                return True
            self.redis.unwatch()
            time.sleep(0.05)  # 재시도 대기

    def release(self):
        """자원 해제"""
        self.redis.decr(self.name)

# Redis 연결 (세마포어 저장소)
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Kafka Consumer 설정
consumer = KafkaConsumer(
    'task_topic',
    bootstrap_servers=['localhost:9092'],
    group_id='processing_group',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

# 세마포어 생성 (예: 동시 처리 가능 Consumer 개수 3개로 제한)
semaphore = DistributedSemaphore(r, "consumer_limit", 3)

def process_message(msg):
    """실제 비즈니스 로직"""
    print(f"[처리 시작] {msg}")
    time.sleep(2)  # 실제 처리 시뮬레이션
    print(f"[처리 완료] {msg}")

# Kafka 메시지 소비 루프
for message in consumer:
    if semaphore.acquire():
        try:
            process_message(message.value)
        finally:
            semaphore.release()
    else:
        print("처리 슬롯이 가득 찼습니다. 잠시 후 재시도.")

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

import threading

BUFFER_SIZE = 3
buffer_semaphore = threading.Semaphore(BUFFER_SIZE)
buffer_lock = threading.Lock()  # 버퍼 보호용 뮤텍스
db_lock = threading.Lock()      # 데이터베이스 보호용 뮤텍스

def use_shared_resources(thread_id):
    buffer_semaphore.acquire()  # 버퍼 접근 제한(동시 3개)
    with buffer_lock:           # 버퍼 내 데이터에 대한 상호배제
        print(f"Thread-{thread_id}: 버퍼 내 데이터 사용")
        with db_lock:           # 데이터베이스 자원 독점
            print(f"Thread-{thread_id}: 데이터베이스 접근")
    buffer_semaphore.release()  # 버퍼 반환

threads = []
for i in range(5):
    t = threading.Thread(target=use_shared_resources, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

  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
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218

const express = require('express');
const axios = require('axios');

class SemaphorePool {
    constructor() {
        this.semaphores = new Map();
    }
    
    // 서비스별 세마포어 생성/조회
    getSemaphore(serviceName, maxConcurrent) {
        if (!this.semaphores.has(serviceName)) {
            this.semaphores.set(serviceName, new Semaphore(maxConcurrent));
        }
        return this.semaphores.get(serviceName);
    }
    
    // 현재 상태 조회 (모니터링용)
    getStatus() {
        const status = {};
        for (const [service, semaphore] of this.semaphores) {
            status[service] = {
                available: semaphore.available,
                total: semaphore.total,
                waiting: semaphore.waitingCount
            };
        }
        return status;
    }
}

class Semaphore {
    constructor(maxConcurrent) {
        this.total = maxConcurrent;
        this.available = maxConcurrent;
        this.waitingQueue = [];
    }
    
    // 비동기 acquire - Promise 기반
    async acquire() {
        return new Promise((resolve) => {
            if (this.available > 0) {
                this.available--;
                resolve();
            } else {
                // 대기 큐에 추가
                this.waitingQueue.push(resolve);
            }
        });
    }
    
    // permit 해제 및 대기 중인 요청 처리
    release() {
        if (this.waitingQueue.length > 0) {
            // 대기 중인 요청을 먼저 처리 (FIFO)
            const nextResolve = this.waitingQueue.shift();
            nextResolve();
        } else {
            this.available++;
        }
    }
    
    get waitingCount() {
        return this.waitingQueue.length;
    }
}

class APIGateway {
    constructor() {
        this.app = express();
        this.semaphorePool = new SemaphorePool();
        
        // 서비스별 설정
        this.serviceConfig = {
            'user-service': { 
                url: 'http://user-service:3001', 
                maxConcurrent: 10,
                timeout: 5000,
                fallback: { error: 'User service unavailable' }
            },
            'movie-service': { 
                url: 'http://movie-service:3002', 
                maxConcurrent: 15,
                timeout: 3000,
                fallback: { movies: [] }
            },
            'rating-service': { 
                url: 'http://rating-service:3003', 
                maxConcurrent: 8,
                timeout: 4000,
                fallback: { rating: 0 }
            }
        };
        
        this.setupRoutes();
    }
    
    setupRoutes() {
        // 동적 프록시 라우팅
        this.app.use('/api/:service/*', async (req, res) => {
            const serviceName = req.params.service;
            const config = this.serviceConfig[serviceName];
            
            if (!config) {
                return res.status(404).json({ error: 'Service not found' });
            }
            
            await this.proxyRequest(serviceName, req, res);
        });
        
        // 모니터링 엔드포인트
        this.app.get('/health/semaphores', (req, res) => {
            res.json({
                timestamp: new Date().toISOString(),
                semaphores: this.semaphorePool.getStatus()
            });
        });
    }
    
    async proxyRequest(serviceName, req, res) {
        const config = this.serviceConfig[serviceName];
        const semaphore = this.semaphorePool.getSemaphore(
            serviceName, 
            config.maxConcurrent
        );
        
        const startTime = Date.now();
        
        try {
            // 세마포어 획득 대기
            await semaphore.acquire();
            console.log(`[${serviceName}] Request acquired permit`);
            
            // 실제 서비스 호출
            const targetUrl = `${config.url}${req.path}`;
            const response = await axios({
                method: req.method,
                url: targetUrl,
                data: req.body,
                headers: {
                    ...req.headers,
                    'x-forwarded-for': req.ip,
                    'x-gateway-time': startTime
                },
                timeout: config.timeout
            });
            
            // 성공 응답 전달
            res.status(response.status).json(response.data);
            
        } catch (error) {
            console.error(`[${serviceName}] Request failed:`, error.message);
            
            // 타임아웃이나 서비스 에러 시 fallback 응답
            if (error.code === 'ECONNABORTED' || error.response?.status >= 500) {
                res.status(503).json({
                    ...config.fallback,
                    _gateway_fallback: true,
                    _error: error.message
                });
            } else {
                res.status(error.response?.status || 500).json({
                    error: error.message,
                    _gateway_error: true
                });
            }
            
        } finally {
            // 세마포어 해제 (반드시 실행)
            semaphore.release();
            
            const duration = Date.now() - startTime;
            console.log(`[${serviceName}] Request completed in ${duration}ms`);
        }
    }
    
    listen(port) {
        this.app.listen(port, () => {
            console.log(`API Gateway listening on port ${port}`);
            console.log('Service configurations:');
            Object.entries(this.serviceConfig).forEach(([name, config]) => {
                console.log(`  ${name}: max ${config.maxConcurrent} concurrent`);
            });
        });
    }
}

// 실행
const gateway = new APIGateway();
gateway.listen(3000);

// 부하 테스트 스크립트 (별도 실행)
async function loadTest() {
    const requests = [];
    
    // 100개의 동시 요청 생성
    for (let i = 0; i < 100; i++) {
        requests.push(
            axios.get(`http://localhost:3000/api/user-service/users/${i}`)
                .catch(err => ({ error: err.message }))
        );
    }
    
    console.log('부하 테스트 시작: 100개 동시 요청');
    const start = Date.now();
    
    const results = await Promise.all(requests);
    
    const duration = Date.now() - start;
    const successful = results.filter(r => !r.error).length;
    const failed = results.length - successful;
    
    console.log(`테스트 완료 (${duration}ms):`);
    console.log(`  성공: ${successful}, 실패: ${failed}`);
    console.log(`  평균 처리율: ${(results.length / duration * 1000).toFixed(2)} req/sec`);
}

// 10초 후 부하 테스트 실행
setTimeout(loadTest, 10000);