카프카 & 아웃박스 패턴으로 자동매매 안정성 개선

build.gradle

@@ -89,6 +89,9 @@ dependencies {
     // redisson
     implementation 'org.redisson:redisson-spring-boot-starter:3.46.0'
 
+    // hibernate types
+    implementation 'com.vladmihalcea:hibernate-types-60:2.21.1'
+
     compileOnly 'org.projectlombok:lombok'
     runtimeOnly 'io.micrometer:micrometer-registry-prometheus'
     annotationProcessor 'org.projectlombok:lombok'

docker-compose.yml

@@ -0,0 +1,38 @@
+version: '3.8'
+
+services:
+  kafka:
+    image: bitnami/kafka:3.7.0
+    container_name: kafka
+    ports:
+      - '9092:9092'         # 내부 Kafka 통신
+      - '19092:19092'       # 외부(Spring 등) 접근
+    environment:
+      - KAFKA_ENABLE_KRAFT=yes
+      - KAFKA_CFG_NODE_ID=1
+      - KAFKA_CFG_PROCESS_ROLES=broker,controller
+      - KAFKA_KRAFT_CLUSTER_ID=kraft-cluster-id-single-node
+      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
+      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:19092
+      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,EXTERNAL://localhost:19092
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
+      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
+      - ALLOW_PLAINTEXT_LISTENER=yes
+      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
+    volumes:
+      - kafka-data:/bitnami/kafka
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    container_name: kafka-ui
+    ports:
+      - "8080:8080"
+    environment:
+      - KAFKA_CLUSTERS_0_NAME=local
+      - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092
+      - KAFKA_CLUSTERS_0_ZOOKEEPER=zookeeper:2181
+    depends_on:
+      - kafka
+
+volumes:
+  kafka-data:

src/main/java/com/tradin/TradinApplication.java

@@ -6,13 +6,15 @@
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 @SpringBootApplication
 @EnableJpaAuditing
 @EnableFeignClients
 @EnableCaching
+@EnableScheduling
 @RestController
 public class TradinApplication {
 

src/main/java/com/tradin/common/aop/AopForTransaction.java

@@ -23,7 +23,7 @@ public Object proceed(final ProceedingJoinPoint joinPoint) throws Throwable {
             );
             throw throwable;
         } finally {
-            log.info("🔓 트랜잭션 종료 - method: {}", joinPoint.getSignature().getName());
+//            log.info("🔓 트랜잭션 종료 - method: {}", joinPoint.getSignature().getName());
         }
     }
 }

src/main/java/com/tradin/common/aop/DistributedLockAop.java

@@ -45,7 +45,7 @@ public Object handleDistributedLock(ProceedingJoinPoint joinPoint, DistributedLo
                 return invokeFallback(joinPoint, distributedLock.fallbackMethod());
             }
 
-            log.info("🔐 락 획득 성공 - key: {}", key);
+//            log.info("🔐 락 획득 성공 - key: {}", key);
             return aopForTransaction.proceed(joinPoint);
 
         } catch (InterruptedException e) {
@@ -56,9 +56,9 @@ public Object handleDistributedLock(ProceedingJoinPoint joinPoint, DistributedLo
             if (isLocked && lock.isHeldByCurrentThread()) {
                 try {
                     lock.unlock();
-                    log.info("🔓 락 해제 - key: {}", key);
+//                    log.info("🔓 락 해제 - key: {}", key);
                 } catch (IllegalMonitorStateException e) {
-                    log.warn("⚠️ 락 이미 해제됨 - method: {}, key: {}", method.getName(), key);
+//                    log.warn("⚠️ 락 이미 해제됨 - method: {}, key: {}", method.getName(), key);
                 }
             }
         }

src/main/java/com/tradin/common/config/AsyncConfiguration.java

@@ -10,8 +10,13 @@
 @EnableAsync
 public class AsyncConfiguration {
 
-    @Bean(name = "tradeExecutor")
+    @Bean(name = "autoTradeExecutor")
     public Executor tradeExecutor() {
         return Executors.newVirtualThreadPerTaskExecutor();
     }
+
+    @Bean(name = "outboxExecutor")
+    public Executor outboxExecutor() {
+        return Executors.newVirtualThreadPerTaskExecutor();
+    }
 }

src/main/java/com/tradin/common/config/KafkaConfiguration.java

@@ -0,0 +1,45 @@
+package com.tradin.common.config;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.core.KafkaAdmin;
+
+@Configuration
+public class KafkaConfiguration {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    public static final String AUTO_TRADE_TOPIC = "auto-trade-topic";
+    public static final String AUTO_TRADE_RETRY_TOPIC = "auto-trade-retry-topic";
+
+
+    @Bean
+    public KafkaAdmin kafkaAdmin() { //TODO
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        return new KafkaAdmin(configs);
+    }
+
+    @Bean
+    public NewTopic autoTradeTopic() {
+        return TopicBuilder.name(AUTO_TRADE_TOPIC)
+            .partitions(3)
+            .replicas(1)
+            .build();
+    }
+
+    @Bean
+    public NewTopic autoTradeRetryTopic() {
+        return TopicBuilder.name(AUTO_TRADE_RETRY_TOPIC)
+            .partitions(1)
+            .replicas(1)
+            .build();
+    }
+} 

src/main/java/com/tradin/common/exception/ExceptionType.java

@@ -43,7 +43,8 @@ public enum ExceptionType {
     NOT_FOUND_HISTORY_EXCEPTION(NOT_FOUND, "존재하지 않는 전략 내역입니다."),
     NOT_FOUND_PRICE_EXCEPTION(NOT_FOUND, "시세가 존재하지 않습니다."),
     NOT_FOUND_BALANCE_EXCEPTION(NOT_FOUND, "잔고가 존재하지 않습니다."),
-    NOT_FOUND_SUCH_METHOD_EXCEPTION(NOT_FOUND, "존재하지 않는 메소드입니다."),
+    NOT_FOUND_SUCH_METHOD_EXCEPTION(NOT_FOUND, "존재하지 않는 메서드입니다."),
+    NOT_FOUND_OUTBOX_EVENT_EXCEPTION(NOT_FOUND, "존재하지 않는 Outbox Event입니다."),
 
     //405 Method Not Allowed
     LOCK_ACQUISITION_FAILED_EXCEPTION(BAD_REQUEST, "락을 획득하지 못했습니다."),
@@ -61,7 +62,11 @@ public enum ExceptionType {
     DECRYPT_FAIL_EXCEPTION(INTERNAL_SERVER_ERROR, "복호화에 실패하였습니다."),
     SIGNATURE_GENERATION_FAIL_EXCEPTION(INTERNAL_SERVER_ERROR, "JWT 서명 생성에 실패하였습니다."),
     PUBLIC_KEY_GENERATE_FAIL_EXCEPTION(INTERNAL_SERVER_ERROR, "공개키 생성에 실패하였습니다."),
-    INTERNAL_SERVER_ERROR_EXCEPTION(INTERNAL_SERVER_ERROR, "서버 내부 오류입니다.");;
+    INTERNAL_SERVER_ERROR_EXCEPTION(INTERNAL_SERVER_ERROR, "서버 내부 오류입니다."),
+    SERIALIZATION_FAIL_EXCEPTION(INTERNAL_SERVER_ERROR, "직렬화에 실패하였습니다."),
+    DESERIALIZATION_FAIL_EXCEPTION(INTERNAL_SERVER_ERROR, "역직렬화에 실패하였습니다."),
+
+    ;
 
     private final HttpStatus httpStatus;
     private final String message;