Distributed Locking

 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

import redis
import time
import uuid

class RedisLock:
    def __init__(self, redis_client, key, ttl=10):
        self.redis = redis_client
        self.key = key
        self.ttl = ttl
        self.id = str(uuid.uuid4())

    def acquire(self, retry_delay=0.01, timeout=5):
        end = time.time() + timeout
        while time.time() < end:
            if self.redis.set(self.key, self.id, nx=True, ex=self.ttl):
                return True
            time.sleep(retry_delay)
        return False

    def release(self):
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        return self.redis.eval(script, 1, self.key, self.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

from kazoo.client import KazooClient
import threading

class ZkLock:
    def __init__(self, zk_hosts, path):
        self.zk = KazooClient(hosts=zk_hosts)
        self.zk.start()
        self.root = path
        self.node = None
        self.cv = threading.Condition()

    def acquire(self):
        self.node = self.zk.create(f"{self.root}/lock-", ephemeral=True, sequence=True)
        while True:
            children = sorted(self.zk.get_children(self.root))
            if self.node.split('/')[-1] == children[0]:
                return True
            prev = children[children.index(self.node.split('/')[-1]) - 1]
            prev_path = f"{self.root}/{prev}"
            event = self.zk.exists(prev_path, watch=lambda e: self._wake())
            with self.cv:
                self.cv.wait()

    def _wake(self):
        with self.cv:
            self.cv.notify()

    def release(self):
        self.zk.delete(self.node)
        self.node = None

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

const { Etcd3 } = require('etcd3');
const client = new Etcd3();

async function acquireLock(key, ttl = 30) {
  const lease = client.lease(ttl);
  const txn = client.if(key, "Create", "=", 0)
    .then(client.put(key).value("locked").lease(lease))
    .commit();
  const res = await txn;
  if (res.succeeded) {
    lease.on('lost', () => console.log('Lease lost'));
    return lease;
  } else {
    await lease.revoke();
    return null;
  }
}

async function releaseLock(lease) {
  if (lease) {
    await lease.revoke();
  }
}

 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

import redis
import time

class DistributedLock:
    def __init__(self, redis_url, lock_key, expiry=5):
        self.r = redis.Redis.from_url(redis_url)
        self.lock_key = lock_key
        self.expiry = expiry  # 락 TTL(초)
        self.token = str(time.time())  # 고유 토큰

    def acquire(self):
        # SET NX PX=5000: 이미 락이 있으면 실패
        result = self.r.set(self.lock_key, self.token, nx=True, ex=self.expiry)
        return result is True

    def release(self):
        # 조건부 DEL: 내 토큰일 때만 락 해제 (다른 인스턴스가 실수로 DEL 방지)
        value = self.r.get(self.lock_key)  # 현재 락 소유자 확인
        if value and value.decode() == self.token:
            self.r.delete(self.lock_key)
            return True
        return False

# 사용 예시
lock = DistributedLock("redis://localhost:6379", "order:lock:123")
if lock.acquire():
    try:
        # 결제 및 주문 로직(단일 실행 보장)
        pass
    finally:
        lock.release()
else:
    print("다른 노드가 락을 선점하여 주문 처리 거절")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from kazoo.client import KazooClient
zk=KazooClient(hosts='127.0.0.1:2181')
zk.start()
lock = zk.Lock("/locks/jobA")
if lock.acquire(timeout=5):
    try:
        run_job()
    finally:
        lock.release()
zk.stop()

  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

import redis
import time
import uuid
from contextlib import contextmanager

class FlashSaleInventoryManager:
    def __init__(self, redis_cluster_nodes):
        self.redis_pools = {}
        for node in redis_cluster_nodes:
            self.redis_pools[node['name']] = redis.ConnectionPool(
                host=node['host'], 
                port=node['port'],
                max_connections=20
            )
        self.redis_clients = {
            name: redis.Redis(connection_pool=pool) 
            for name, pool in self.redis_pools.items()
        }
        
    @contextmanager
    def acquire_product_lock(self, product_id, timeout=5):
        """제품별 분산 락 컨텍스트 매니저"""
        lock_key = f"flash_sale_lock:product:{product_id}"
        lock_value = str(uuid.uuid4())
        ttl_ms = timeout * 1000
        
        # Redlock 알고리즘 구현
        acquired_locks = []
        start_time = time.time() * 1000
        
        try:
            # 모든 Redis 노드에서 락 획득 시도
            for name, client in self.redis_clients.items():
                try:
                    if client.set(lock_key, lock_value, nx=True, px=ttl_ms):
                        acquired_locks.append((name, client))
                except redis.RedisError:
                    continue
            
            elapsed_time = (time.time() * 1000) - start_time
            quorum = len(self.redis_clients) // 2 + 1
            
            # 과반수 락 획득 및 시간 검증
            if len(acquired_locks) >= quorum and elapsed_time < ttl_ms:
                remaining_ttl = ttl_ms - elapsed_time
                yield lock_value, remaining_ttl
            else:
                raise Exception(f"락 획득 실패: {product_id}")
                
        finally:
            # 락 해제 (Lua 스크립트로 안전한 해제)
            release_script = """
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("del", KEYS[1])
            else
                return 0
            end
            """
            
            for name, client in acquired_locks:
                try:
                    client.eval(release_script, 1, lock_key, lock_value)
                except redis.RedisError:
                    continue

    def process_flash_sale_order(self, product_id, user_id, quantity):
        """플래시 세일 주문 처리"""
        
        # 분산 락으로 동시성 제어
        with self.acquire_product_lock(product_id, timeout=3) as (lock_token, ttl):
            
            # 현재 재고 확인
            current_inventory = self.get_current_inventory(product_id)
            
            if current_inventory < quantity:
                return {
                    "success": False,
                    "message": "재고 부족",
                    "available": current_inventory
                }
            
            # 재고 차감 및 주문 생성 (원자적 트랜잭션)
            order_id = self.create_order_transaction(
                product_id, user_id, quantity, lock_token
            )
            
            if order_id:
                # 성공 로그 및 모니터링 메트릭
                self.log_successful_order(product_id, user_id, quantity, lock_token)
                
                return {
                    "success": True,
                    "order_id": order_id,
                    "message": "주문 성공"
                }
            else:
                return {
                    "success": False,
                    "message": "주문 처리 실패"
                }

    def get_current_inventory(self, product_id):
        """현재 재고 조회"""
        # MySQL에서 재고 조회
        with get_db_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT quantity FROM inventory WHERE product_id = %s FOR UPDATE",
                (product_id,)
            )
            result = cursor.fetchone()
            return result[0] if result else 0

    def create_order_transaction(self, product_id, user_id, quantity, fencing_token):
        """펜싱 토큰을 포함한 주문 트랜잭션"""
        with get_db_connection() as conn:
            try:
                conn.autocommit = False
                cursor = conn.cursor()
                
                # 펜싱 토큰 검증 (이전 토큰보다 새로운지 확인)
                cursor.execute(
                    "SELECT last_fencing_token FROM product_locks WHERE product_id = %s",
                    (product_id,)
                )
                last_token = cursor.fetchone()
                
                if last_token and last_token[0] >= fencing_token:
                    conn.rollback()
                    return None
                
                # 재고 차감
                cursor.execute(
                    "UPDATE inventory SET quantity = quantity - %s WHERE product_id = %s AND quantity >= %s",
                    (quantity, product_id, quantity)
                )
                
                if cursor.rowcount == 0:
                    conn.rollback()
                    return None
                
                # 주문 생성
                order_id = str(uuid.uuid4())
                cursor.execute(
                    "INSERT INTO orders (order_id, product_id, user_id, quantity, fencing_token, created_at) VALUES (%s, %s, %s, %s, %s, NOW())",
                    (order_id, product_id, user_id, quantity, fencing_token)
                )
                
                # 펜싱 토큰 업데이트
                cursor.execute(
                    "UPDATE product_locks SET last_fencing_token = %s WHERE product_id = %s",
                    (fencing_token, product_id)
                )
                
                conn.commit()
                return order_id
                
            except Exception as e:
                conn.rollback()
                raise e

    def log_successful_order(self, product_id, user_id, quantity, token):
        """성공 주문 로깅 및 메트릭"""
        import logging
        
        logger = logging.getLogger('flash_sale')
        logger.info(f"주문 성공: product={product_id}, user={user_id}, qty={quantity}, token={token}")
        
        # 메트릭 전송 (예: Prometheus)
        # metrics.increment('flash_sale_orders_total', tags={'product_id': product_id})

# 사용 예시
def handle_flash_sale_request(product_id, user_id, quantity):
    """플래시 세일 요청 처리 핸들러"""
    
    redis_nodes = [
        {'name': 'redis1', 'host': 'redis1.internal', 'port': 6379},
        {'name': 'redis2', 'host': 'redis2.internal', 'port': 6379},
        {'name': 'redis3', 'host': 'redis3.internal', 'port': 6379}
    ]
    
    inventory_manager = FlashSaleInventoryManager(redis_nodes)
    
    try:
        result = inventory_manager.process_flash_sale_order(
            product_id, user_id, quantity
        )
        return result
        
    except Exception as e:
        return {
            "success": False,
            "message": f"시스템 오류: {str(e)}"
        }

 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 redis
import time

def acquire_lock(redis_client, lock_key, timeout=5):
    expire_time = int(time.time()) + timeout
    if redis_client.set(lock_key, expire_time, nx=True, ex=timeout):
        # 락 획득 성공
        return True
    return False

def release_lock(redis_client, lock_key):
    redis_client.delete(lock_key) # 락 해제

# 실사용 예시
r = redis.StrictRedis(host='localhost', port=6379)
if acquire_lock(r, "order:lock:1234"):
    try:
        # 결제 로직
        pass
    finally:
        release_lock(r, "order:lock:1234")
else:
    # 이미 처리 중이므로 중복 방지
    print("이미 결제 처리 중입니다.")

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

from kazoo.client import KazooClient
from kazoo.recipe.lock import Lock

zk = KazooClient(hosts='zk1:2181,zk2:2181,zk3:2181')
zk.start()

lock = Lock(zk, '/locks/batch_job_A')

if lock.acquire(timeout=5):  # TTL 또는 timeout 설정
    try:
        print("Lock acquired, running job…")
        run_batch_job()  # 작업 실행
    finally:
        lock.release()
        print("Lock released.")
else:
    print("Lock not acquired, skipping execution.")

zk.stop()

  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

import redis
import time
import uuid
import logging
from contextlib import contextmanager
from typing import Optional

class InventoryManager:
    """분산 잠금을 활용한 재고 관리 시스템"""
    
    def __init__(self, redis_client, db_connection):
        self.redis = redis_client
        self.db = db_connection
        self.logger = logging.getLogger(__name__)
    
    @contextmanager
    def distributed_lock(self, resource_id: str, timeout: int = 10):
        """분산 잠금 컨텍스트 매니저"""
        lock_key = f"inventory_lock:{resource_id}"
        identifier = str(uuid.uuid4())
        
        # 잠금 획득 시도
        acquired = False
        try:
            acquired = self._acquire_lock(lock_key, identifier, timeout)
            if not acquired:
                raise TimeoutError(f"Failed to acquire lock for {resource_id}")
            
            self.logger.info(f"Lock acquired for {resource_id} with ID {identifier}")
            yield identifier
            
        finally:
            if acquired:
                self._release_lock(lock_key, identifier)
                self.logger.info(f"Lock released for {resource_id}")
    
    def _acquire_lock(self, lock_key: str, identifier: str, timeout: int) -> bool:
        """Redis를 사용한 잠금 획득"""
        end_time = time.time() + timeout
        
        while time.time() < end_time:
            # 30초 TTL로 잠금 설정
            if self.redis.set(lock_key, identifier, nx=True, ex=30):
                return True
            time.sleep(0.001)  # 1ms 대기
        
        return False
    
    def _release_lock(self, lock_key: str, identifier: str) -> bool:
        """안전한 잠금 해제"""
        # Lua 스크립트로 원자적 해제
        release_script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        return bool(self.redis.eval(release_script, 1, lock_key, identifier))
    
    def process_order(self, product_id: str, quantity: int, customer_id: str) -> dict:
        """주문 처리 (분산 잠금 활용)"""
        try:
            # 제품별 분산 잠금 획득
            with self.distributed_lock(product_id, timeout=5) as lock_id:
                # 현재 재고 확인
                current_stock = self._get_current_stock(product_id)
                
                if current_stock < quantity:
                    return {
                        "success": False,
                        "message": f"Insufficient stock. Available: {current_stock}",
                        "lock_id": lock_id
                    }
                
                # 재고 차감 (원자적 연산)
                updated_stock = self._update_stock(product_id, -quantity)
                
                # 주문 기록 생성
                order_id = self._create_order(product_id, quantity, customer_id)
                
                # 성공 로그 기록
                self.logger.info(
                    f"Order processed successfully: "
                    f"order_id={order_id}, product_id={product_id}, "
                    f"quantity={quantity}, remaining_stock={updated_stock}"
                )
                
                return {
                    "success": True,
                    "order_id": order_id,
                    "remaining_stock": updated_stock,
                    "lock_id": lock_id
                }
                
        except TimeoutError as e:
            self.logger.error(f"Lock timeout for product {product_id}: {e}")
            return {"success": False, "message": "System busy, please try again"}
        
        except Exception as e:
            self.logger.error(f"Order processing failed: {e}")
            return {"success": False, "message": "Internal error"}
    
    def _get_current_stock(self, product_id: str) -> int:
        """현재 재고 조회"""
        cursor = self.db.cursor()
        cursor.execute(
            "SELECT stock_quantity FROM inventory WHERE product_id = %s",
            (product_id,)
        )
        result = cursor.fetchone()
        return result[0] if result else 0
    
    def _update_stock(self, product_id: str, delta: int) -> int:
        """재고 업데이트"""
        cursor = self.db.cursor()
        cursor.execute(
            """
            UPDATE inventory 
            SET stock_quantity = stock_quantity + %s,
                last_updated = NOW()
            WHERE product_id = %s
            RETURNING stock_quantity
            """,
            (delta, product_id)
        )
        result = cursor.fetchone()
        self.db.commit()
        return result[0] if result else 0
    
    def _create_order(self, product_id: str, quantity: int, customer_id: str) -> str:
        """주문 생성"""
        order_id = f"ORD-{int(time.time())}-{uuid.uuid4().hex[:8]}"
        cursor = self.db.cursor()
        cursor.execute(
            """
            INSERT INTO orders (order_id, product_id, quantity, customer_id, created_at)
            VALUES (%s, %s, %s, %s, NOW())
            """,
            (order_id, product_id, quantity, customer_id)
        )
        self.db.commit()
        return order_id

# 사용 예시
if __name__ == "__main__":
    # Redis 및 DB 연결 설정
    redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
    db_connection = get_database_connection()  # 가상의 DB 연결 함수
    
    # 재고 관리자 초기화
    inventory_manager = InventoryManager(redis_client, db_connection)
    
    # 동시 주문 시나리오 테스트
    result1 = inventory_manager.process_order("PHONE-123", 1, "customer-1")
    result2 = inventory_manager.process_order("PHONE-123", 1, "customer-2")
    
    print("Order 1 Result:", result1)
    print("Order 2 Result:", result2)

 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

import time, uuid
import redis

redis_cli = redis.Redis()
lock_key = "lock:item:123"

def acquire_lock(timeout=10):
    token = str(uuid.uuid4())
    if redis_cli.set(lock_key, token, nx=True, ex=timeout):
        return token
    return None

def release_lock(token):
    script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """
    redis_cli.eval(script, 1, lock_key, token)

# 사용 예시
token = acquire_lock()
if token:
    try:
        # 공유 자원 작업
        pass
    finally:
        release_lock(token)

  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
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268

import redis
import time
import uuid
import threading
from typing import Optional, Any
import logging

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DistributedLock:
    """Redis 기반 분산 락 구현"""
    
    def __init__(self, redis_client: redis.Redis, lock_key: str, 
                 timeout: int = 30, blocking_timeout: int = 10):
        """
        분산 락 초기화
        
        Args:
            redis_client: Redis 클라이언트 인스턴스
            lock_key: 락의 고유 키
            timeout: 락 유지 시간 (초)
            blocking_timeout: 락 획득 대기 시간 (초)
        """
        self.redis = redis_client
        self.lock_key = f"lock:{lock_key}"
        self.timeout = timeout
        self.blocking_timeout = blocking_timeout
        self.identifier = str(uuid.uuid4())  # 락 소유자 식별자
        self.renewal_thread = None
        self.stop_renewal = False
    
    def acquire(self) -> bool:
        """
        락 획득 시도
        
        Returns:
            bool: 락 획득 성공 여부
        """
        end_time = time.time() + self.blocking_timeout
        
        while time.time() < end_time:
            # SET NX (Not Exists) + EX (Expire) 를 사용한 원자적 락 획득
            if self.redis.set(self.lock_key, self.identifier, 
                             nx=True, ex=self.timeout):
                logger.info(f"Lock acquired: {self.lock_key}")
                
                # 락 갱신 스레드 시작
                self._start_renewal_thread()
                return True
            
            # 짧은 대기 후 재시도
            time.sleep(0.01)
        
        logger.warning(f"Failed to acquire lock: {self.lock_key}")
        return False
    
    def release(self) -> bool:
        """
        락 해제
        
        Returns:
            bool: 락 해제 성공 여부
        """
        # 락 갱신 중지
        self.stop_renewal = True
        if self.renewal_thread:
            self.renewal_thread.join()
        
        # Lua 스크립트를 사용한 원자적 락 해제
        # 자신이 소유한 락만 해제 가능
        lua_script = """
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
        """
        
        result = self.redis.eval(lua_script, 1, self.lock_key, self.identifier)
        
        if result:
            logger.info(f"Lock released: {self.lock_key}")
            return True
        else:
            logger.warning(f"Failed to release lock (not owner): {self.lock_key}")
            return False
    
    def _start_renewal_thread(self):
        """락 갱신 스레드 시작"""
        self.stop_renewal = False
        self.renewal_thread = threading.Thread(target=self._renew_lock)
        self.renewal_thread.daemon = True
        self.renewal_thread.start()
    
    def _renew_lock(self):
        """
        락 자동 갱신 (TTL의 1/3 주기로 갱신)
        """
        renewal_interval = self.timeout // 3
        
        while not self.stop_renewal:
            time.sleep(renewal_interval)
            
            if self.stop_renewal:
                break
            
            # 락이 여전히 자신의 소유인지 확인하고 갱신
            lua_script = """
            if redis.call("GET", KEYS[1]) == ARGV[1] then
                return redis.call("EXPIRE", KEYS[1], ARGV[2])
            else
                return 0
            end
            """
            
            result = self.redis.eval(lua_script, 1, self.lock_key, 
                                   self.identifier, self.timeout)
            
            if result:
                logger.debug(f"Lock renewed: {self.lock_key}")
            else:
                logger.warning(f"Lock renewal failed: {self.lock_key}")
                break
    
    def __enter__(self):
        """Context Manager 진입"""
        if not self.acquire():
            raise RuntimeError(f"Failed to acquire lock: {self.lock_key}")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context Manager 종료"""
        self.release()

class OrderService:
    """주문 서비스 - 분산 락을 활용한 재고 관리"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.inventory = {}  # 간단한 메모리 재고 관리
        
        # 초기 재고 설정
        self.inventory = {
            "product_001": 100,
            "product_002": 50,
            "product_003": 25
        }
    
    def process_order(self, product_id: str, quantity: int, 
                     customer_id: str) -> dict:
        """
        주문 처리 - 분산 락을 사용한 안전한 재고 차감
        
        Args:
            product_id: 상품 ID
            quantity: 주문 수량
            customer_id: 고객 ID
            
        Returns:
            dict: 주문 처리 결과
        """
        # 상품별 락 키 생성
        lock_key = f"inventory:{product_id}"
        
        try:
            # 분산 락을 사용한 재고 처리
            with DistributedLock(self.redis, lock_key, timeout=30):
                logger.info(f"Processing order: {product_id}, "
                          f"quantity: {quantity}, customer: {customer_id}")
                
                # 현재 재고 확인
                current_stock = self.inventory.get(product_id, 0)
                
                if current_stock < quantity:
                    logger.warning(f"Insufficient stock: {product_id}, "
                                 f"available: {current_stock}, requested: {quantity}")
                    return {
                        "success": False,
                        "message": "재고 부족",
                        "available_stock": current_stock
                    }
                
                # 재고 차감 (실제로는 데이터베이스 업데이트)
                self.inventory[product_id] -= quantity
                
                # 주문 생성 로직 시뮬레이션
                time.sleep(0.1)  # 데이터베이스 작업 시뮬레이션
                
                order_id = f"ORDER_{int(time.time())}_{customer_id}"
                
                logger.info(f"Order completed: {order_id}, "
                          f"remaining stock: {self.inventory[product_id]}")
                
                return {
                    "success": True,
                    "order_id": order_id,
                    "product_id": product_id,
                    "quantity": quantity,
                    "remaining_stock": self.inventory[product_id]
                }
                
        except RuntimeError as e:
            logger.error(f"Failed to acquire lock for {product_id}: {e}")
            return {
                "success": False,
                "message": "시스템 오류 - 락 획득 실패"
            }
        except Exception as e:
            logger.error(f"Order processing error: {e}")
            return {
                "success": False,
                "message": f"주문 처리 오류: {str(e)}"
            }
    
    def get_inventory_status(self) -> dict:
        """현재 재고 상태 조회"""
        return self.inventory.copy()

# 사용 예시 및 테스트
def demo_distributed_locking():
    """분산 락킹 데모"""
    print("=== Distributed Locking Demo ===\n")
    
    # Redis 연결
    redis_client = redis.Redis(host='localhost', port=6379, db=0, 
                              decode_responses=True)
    
    # 주문 서비스 생성
    order_service = OrderService(redis_client)
    
    print("초기 재고 상태:")
    print(order_service.get_inventory_status())
    print()
    
    # 동시 주문 시뮬레이션
    def simulate_order(product_id: str, quantity: int, customer_id: str):
        """주문 시뮬레이션 함수"""
        result = order_service.process_order(product_id, quantity, customer_id)
        print(f"Customer {customer_id}: {result}")
    
    # 여러 스레드로 동시 주문 테스트
    threads = []
    
    # 동일 상품에 대한 동시 주문
    for i in range(5):
        thread = threading.Thread(
            target=simulate_order,
            args=("product_001", 20, f"CUST_{i:03d}")
        )
        threads.append(thread)
    
    print("동시 주문 처리 시작…")
    
    # 모든 스레드 시작
    for thread in threads:
        thread.start()
    
    # 모든 스레드 완료 대기
    for thread in threads:
        thread.join()
    
    print("\n최종 재고 상태:")
    print(order_service.get_inventory_status())

if __name__ == "__main__":
    demo_distributed_locking()