You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 21 Next »

액터모델의 메일박스로 통해 전달되는 이벤트들은 다양한 DB장치를 이용해 영속화가 가능하며 상태프로그래밍을 할수 있습니다.
RDB,MongoDB,ElasticSearch등 다양한 특징을 가진 전통강호 영속장치를 이용해 각각 특화된 기능을 이용할수 있겠지만
이러한 기능을 한방에 보편적으로 이용할수 있는 장치가 없을까 조사하다 발견한 DB로 Akka를 연구하는 NetCorelabs에 변종실험 대상으로 추가하였습니다.

데이터베이스는 지루하면 안된다라는 슬로건을 내건 모던 DB

✅ RavenDB 특징

항목설명
Document StoreJSON 기반의 문서 저장 (MongoDB처럼)
Full-Text Search 내장Lucene 기반 검색엔진 포함 (Elasticsearch 대체 가능)
Graph-Like Traversal 지원Include, Load, RelatedDocuments 로 Graph traversal 흉내 가능
벡터 검색 (Vector Search)6.0 이상 버전에서 Vector search 지원 (Preview → Stable 예정)
ACID 트랜잭션 지원NoSQL 중 드물게 단일 DB 내 ACID 지원
자동 인덱싱/쿼리 최적화쿼리 기반으로 자동 인덱싱 생성
Change Vector / ETL 기능 내장다른 Raven 클러스터 또는 외부 시스템으로 데이터 복제 가능
클라우드 + 온프렘 지원다양한 배포 환경 대응
Sharding + Replication분산 구조 대응 가능 (Sharded DB)


✅ 기존 DB 구성 중 대체 가능한 역할


기존 시스템RavenDB로 대체 가능 여부설명
MongoDB (Document DB)✅ 완전 대체JSON 기반 문서 저장, 컬렉션 → 문서 분리 모델
Elasticsearch✅ 부분 대체Full-text 검색 지원, 복잡한 분석쿼리는 제한적이나 일반 검색에는 충분
Neo4j (Graph DB)⚠️ 간단한 관계 트래버설은 가능명시적 Graph 모델링은 어려움 (복잡한 네트워크 분석에는 부적합)
Vector DB (예: Weaviate, Milvus)✅ 단순 벡터 검색은 대체 가능다차원 벡터 검색 API 제공, 모델링+쿼리 결합 쉬움
RDB (CRUD/정형)⚠️ 단순 CRUD는 가능, 복잡한 조인과 트랜잭션은 제한적정형 테이블 기반보다는 문서 중심 모델 필요


RavenDB Docker StandAlone 구동

version: '3.8'

services:
  ravendb:
    image: ravendb/ravendb:ubuntu-latest
    container_name: ravendb
    ports:
      - "9000:8080"
    environment:
      - RAVEN_Setup_Mode=None
      - RAVEN_License_Eula_Accepted=true
    volumes:
      - ravendb_data:/ravendb/data
      - ravendb_logs:/ravendb/logs

volumes:
  ravendb_data:
  ravendb_logs:


  • 클라우드로도 이용가능하며~ 로컬또는 온프레미스로도 운영가능합니다. 


IDE 환경

  • Docker와 통합된 IDE환경으로 RavenDB 구동
  • 닷넷개발 IDE VisualStudio가 개인적으로 익숙하지만 최근 젯브레인진영 Rider를 이용해보고 있습니다. 

RavenDB 웹관리툴

  • DB를 관리할 웹툴을 포함하며 기본적인 관리가 가능합니다.
  • IDE내에서 연동돠는 플러그인은 아직 없어보입니다.
    • Postgres 호환플러그인을 지원하니 BI툴및 IDE내 기본조회연동은 가능할듯


언어별 지원 클라이언트

  • Akka.net을  연계해 사용예정이기때문에 .NET 기반실험이 진행되었습니다.
  • 지원 클라이언트가 소개될때 .NET 라이브러리가 첫번째에 잘 노출되는 일이 없는데~ 닷넷친화적 DB로 추정해봅니다.


RavenClient for .NET

  • 설치된 서버 Mazor버전과 맞춰서 패키지 설치


Database생성


몽고DB와 유사하게 스키마리스 DB이기때문에 DDL코드가 필요로 하지 않습니다. 


Repository 코드

public class Member
{
    public string Id { get; set; } // RavenDB는 기본적으로 Id를 문서 키로 사용
    public string Name { get; set; }
    public string Email { get; set; }
    public int Age { get; set; }
}

public class MemberRepository
{
    private readonly IDocumentStore _store;

    public MemberRepository(IDocumentStore store)
    {
        _store = store ?? throw new ArgumentNullException(nameof(store));
    }

    public void AddMember(Member member)
    {
        using (var session = _store.OpenSession())
        {
            session.Store(member);
            session.SaveChanges();
        }
    }

    public Member GetMemberById(string id)
    {
        using (var session = _store.OpenSession())
        {
            return session.Load<Member>(id);
        }
    }

    public void UpdateMember(Member member)
    {
        using (var session = _store.OpenSession())
        {
            var existingMember = session.Load<Member>(member.Id);
            if (existingMember != null)
            {
                existingMember.Name = member.Name;
                existingMember.Email = member.Email;
                existingMember.Age = member.Age;
                session.SaveChanges();
            }
        }
    }

    public void DeleteMember(string id)
    {
        using (var session = _store.OpenSession())
        {
            var member = session.Load<Member>(id);
            if (member != null)
            {
                session.Delete(member);
                session.SaveChanges();
            }
        }
    }
}


CRUD TEST 코드

public class MemberRepositoryTest : TestKitXunit
{
    private readonly IDocumentStore _store;
    private readonly MemberRepository _repository;
    
    public MemberRepositoryTest(ITestOutputHelper output) : base(output)
    {
        // RavenDB 임베디드 서버 초기화
        _store = new DocumentStore
        {
            Urls = new[] { "http://localhost:9000" }, // 로컬 RavenDB URL
            Database = "net-core-labs"
        };
        _store.Initialize();

        // MemberRepository 초기화
        _repository = new MemberRepository(_store);
    }
    
    [Fact]
    public void AddMember_ShouldAddMemberSuccessfully()
    {
        // Arrange
        var member = new Member
        {
            Name = "John Doe",
            Email = "john.doe@example.com",
            Age = 30
        };

        // Act
        _repository.AddMember(member);

        // Assert
        var retrievedMember = _repository.GetMemberById(member.Id);
        Assert.NotNull(retrievedMember);
        Assert.Equal("John Doe", retrievedMember.Name);
    }

    [Fact]
    public void UpdateMember_ShouldUpdateMemberSuccessfully()
    {
        // Arrange
        var member = new Member
        {
            Name = "Jane Doe",
            Email = "jane.doe@example.com",
            Age = 25
        };
        _repository.AddMember(member);

        // Act
        member.Age = 26;
        _repository.UpdateMember(member);

        // Assert
        var updatedMember = _repository.GetMemberById(member.Id);
        Assert.NotNull(updatedMember);
        Assert.Equal(26, updatedMember.Age);
    }

    [Fact]
    public void DeleteMember_ShouldDeleteMemberSuccessfully()
    {
        // Arrange
        var member = new Member
        {
            Name = "Mark Smith",
            Email = "mark.smith@example.com",
            Age = 40
        };
        _repository.AddMember(member);

        // Act
        _repository.DeleteMember(member.Id);

        // Assert
        var deletedMember = _repository.GetMemberById(member.Id);
        Assert.Null(deletedMember);
    }
    
}
  • 처음 사용하는 DB사용 자체를 연구하고 테스트하는 코드이기때문에 Mocking을 이용하지 않고 로컬구축 DB직접 Access방식이용


테스트 수행및 확인

  • 비교적 심플한 코드작성으로 CRUD 테스트 수행이 완료되었습니다.


CQRS로의 여정

CRUD만 이용하려고 RavenDB를 선택한것은 아니고~ Akka.net의 Persist기능과 연동되어 

이벤트 소싱을 액터모델의 Persitent 기능과 함께 심플하게 이용할수 있기때문입니다.


RavenDB가 Akka.net의 Persitence 기능을 지원하며 이 코드를 이해하기전 이벤트 소싱패턴을 액터모델을 함께 이용했을때 특징을 먼저 살펴보고 작동가능한 구현된 샘플코드도 유닛테스트를 통해 살펴보겠습니다.


Akka.NET의 액터모델은 CQRS (Command Query Responsibility Segregation) 및 이벤트 소싱 (Event Sourcing) 패턴을 구현하기에 매우 적합한 구조를 제공합니다. 특히 Akka.Persistence 모듈과의 연계를 통해 상태 저장과 복구, 그리고 이벤트 기반 모델링이 가능해집니다. 아래에 그 기능과 장점을 정리해드립니다.

✅ Akka.NET 액터모델 + Akka.Persistence 를 활용한 CQRS + 이벤트 소싱

1. 핵심 구성요소


구성요소설명
PersistentActor이벤트를 저장하고 재생할 수 있는 상태 유지 액터
Snapshotting빠른 복구를 위해 특정 시점의 상태를 저장
Event Journal모든 상태 변화(event)를 append-only 로그로 저장
Read Model Actor쿼리에 최적화된 projection을 담당
Command Handler Actor명령(Command)을 받아 이벤트로 전환 및 persistence 수행


2. CQRS(이벤트 소싱) with ActorModel

  • 이벤트가 발생할때 즉각 CRUD하여 이용하는 패턴이 아닌 액터모델의 메일박스를 이용해 상태있는 프로그래밍을 통해 이벤트소싱을 설계할수 있으며~ 이러한 느낌


3. 장점 요약

✅ 액터모델 기반의 장점

  • 상태 격리: 각 액터는 고유 상태를 가지며 병렬 처리에 유리함.

  • 비동기 메시지 기반 처리: 병목 없이 고성능 분산 처리 가능.

  • 자연스러운 도메인 분리: 액터 자체가 DDD의 Aggregate Root 역할을 하기에 적합.

✅ CQRS & Event Sourcing의 장점

  • 이벤트 이력 관리: 모든 상태 변화가 이벤트로 저장되므로 과거 상태 추적 가능 (감사, 재처리).

  • 읽기/쓰기 분리: 쿼리와 명령이 분리되어 성능 최적화 가능.

  • 스냅샷 사용 가능: 복구 속도 향상.

  • 스케일 아웃 용이: 각 액터가 독립적으로 배포/스케일링 가능.

✅ Akka.NET 전용의 추가 장점

  • Akka.Cluster와 연계 시 분산 시스템 구성도 가능

  • Akka.Remote로 마이크로서비스 간 통신에도 활용 가능

  • Pluggable Journal Backend: SQL, NoSQL, 이벤트 스토어 등 다양한 저장소와 연동 가능

✨ 실사용 시나리오 예

  • 고객 주문 시스템: OrderActor가 PlaceOrderCommand를 받고 OrderPlacedEvent를 생성 및 저장

  • 챗봇 세션 관리: SessionActor가 메시지를 이벤트로 저장하여 상태 기반 대화 흐름 관리

  • 금융 거래 기록: AccountActor가 Withdrawn, Deposited 이벤트를 기록하여 완전한 거래 이력 확보



SalesActor

public class SalesActor: ReceivePersistentActor
{
    // The unique actor id
    public override string PersistenceId => "sales-actor";
    
    // The state that will be persisted in SNAPSHOTS
    private SalesActorState _state;
    
    public SalesActor(long expectedProfit, TaskCompletionSource<bool> taskCompletion)
    {
        _state = new SalesActorState
        {
            totalSales = 0
        }; 
        
        // Process a sale:
        Command<Sale>(saleInfo =>
        {
            if (_state.totalSales < expectedProfit)
            {
                // Persist an EVENT to RavenDB
                // ===========================
                
                // The handler function is executed after the EVENT was saved successfully
                Persist(saleInfo, _ =>
                {
                    // Update the latest state in the actor
                    _state.totalSales += saleInfo.Price;

                    ConsoleHelper.WriteToConsole(ConsoleColor.Black,
                        $"Sale was persisted. Phone brand: {saleInfo.Brand}. Price: {saleInfo.Price}");

                    // Store a SNAPSHOT every 5 sale events
                    // ====================================
                    
                    if (LastSequenceNr != 0 && LastSequenceNr % 5 == 0)
                    {
                        SaveSnapshot(_state.totalSales);
                    }
                });
            }
            else if (!taskCompletion.Task.IsCompleted)
            {
                Sender.Tell(new StopSimulate());
                
                ConsoleHelper.WriteToConsole(ConsoleColor.DarkMagenta,
                    $"Sale not persisted: " +
                    $"Total sales have already reached the expected profit of {expectedProfit}");
                
                ConsoleHelper.WriteToConsole(ConsoleColor.DarkMagenta,
                    _state.ToString());
                
                taskCompletion.TrySetResult(true);
            }
        });
        
        // Handle a SNAPSHOT success msg
        Command<SaveSnapshotSuccess>(success =>
        {
            ConsoleHelper.WriteToConsole(ConsoleColor.Blue,
                $"Snapshot saved successfully at sequence number {success.Metadata.SequenceNr}");
            
            // Optionally, delete old snapshots or events here if needed
            // DeleteMessages(success.Metadata.SequenceNr);
        });
        
        // Recover an EVENT
        Recover<Sale>(saleInfo =>
        {
            _state.totalSales += saleInfo.Price;
            
            ConsoleHelper.WriteToConsole(ConsoleColor.DarkGreen,
                $"Event was recovered. Price: {saleInfo.Price}");
        });
        
        // Recover a SNAPSHOT
        Recover<SnapshotOffer>(offer =>
        {
            var salesFromSnapshot = (long) offer.Snapshot;
            _state.totalSales = salesFromSnapshot;
            
            ConsoleHelper.WriteToConsole(ConsoleColor.DarkGreen,
                $"Snapshot was recovered. Total sales from snapshot: {salesFromSnapshot}");
        });
    }
}

이 SalesActor 클래스는 **Akka.NET의 ReceivePersistentActor**를 상속하여 **이벤트 소싱(Event Sourcing)**과 스냅샷(Snapshot) 기능을 활용해 상태를 안정적으로 유지하는 **영속 액터(Persistent Actor)**입니다. 아래에 코드 전체 흐름과 구성 요소를 설명드리겠습니다.

🔍 전체 구조 분석

📌 클래스 목적

  • 판매 이벤트(Sale)를 처리하고 누적된 매출을 상태로 유지

  • 특정 조건(예: 매출 목표 달성)에 도달하면 처리 중단

  • RavenDB 등을 백엔드로 사용해 이벤트를 저장

  • 일정 횟수마다 스냅샷 저장하여 복구 최적화

🧱 주요 구성 요소 분석

1. PersistenceId

public override string PersistenceId => "sales-actor";


  • 액터의 고유 식별자로 이벤트와 스냅샷을 식별하기 위한 키

  • RavenDB 또는 다른 Akka.Persistence backend에서 이 키로 데이터를 저장/조회함


2. 상태 관리 변수

private SalesActorState _state;


  • 현재 액터 상태를 저장하는 내부 객체

  • 여기서는 totalSales (누적 매출액)를 관리

3. 명령 처리 (Command<Sale>)

Command<Sale>(saleInfo => { ... });

  • 외부에서 Sale 메시지를 받으면 처리

  • 조건에 따라 이벤트를 Persist하거나 무시


주요 로직

Persist(saleInfo, _ =>
{
    _state.totalSales += saleInfo.Price;
    
    if (LastSequenceNr % 5 == 0) {
        SaveSnapshot(_state.totalSales);
    }
});

  • Persist()는 이벤트 소싱의 핵심: 이벤트를 저장한 후 상태를 변경

  • 5개의 Sale 이벤트마다 SaveSnapshot() 호출 → 빠른 복구용


4. 종료 시그널 전달

Sender.Tell(new StopSimulate());
taskCompletion.TrySetResult(true);

  • 매출 목표(expectedProfit)에 도달하면, 더 이상 이벤트를 저장하지 않고 종료 시그널 전송

  • TaskCompletionSource는 테스트나 외부 시뮬레이션 종료에 사용


5. 스냅샷 성공 처리

Command<SaveSnapshotSuccess>(success => {
    Console.WriteLine(...);
});

  • 스냅샷이 성공적으로 저장되었을 때 로깅

  • 필요시 DeleteMessages() 같은 클린업 가능


6. 복구 처리 (Recover<>)

이벤트 복구

Recover<Sale>(saleInfo => {
    _state.totalSales += saleInfo.Price;
});

스냅샷 복구

Recover<SnapshotOffer>(offer => {
    _state.totalSales = (long)offer.Snapshot;
});

가장 최근의 스냅샷부터 시작 → 남은 이벤트만 재생

✅ 핵심 특징 요약


기능설명
🎯 명령-이벤트 분리Sale은 명령(Command), Persist된 이벤트로 상태 변경
🧠 이벤트 소싱 기반 상태상태 변경 내역은 전부 이벤트로 기록되어 리플레이 가능
💾 스냅샷매 5건 이벤트마다 현재 상태 저장으로 복구 최적화
🔄 복구 내장스냅샷 → 이벤트 순으로 복원함
🚦 목표 달성시 종료외부 TaskCompletionSource로 흐름 제어



유닛테스트를 통한 시뮬레이터

public class SalesSimulatorActor : ReceiveActor
{
    private readonly IActorRef _salesActor;
    private ICancelable scheduler;

    public SalesSimulatorActor(IActorRef salesActor)
    {
        _salesActor = salesActor;

        // Schedule the first sale simulation immediately and then every 2 seconds:
        scheduler = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(TimeSpan.Zero, 
            TimeSpan.FromSeconds(2), Self, new StartSimulate(), Self);
        
        Receive<StartSimulate>(HandleStart);
        Receive<StopSimulate>(HandleStop);
    }

    private void HandleStart(StartSimulate message)
    {
        ConsoleHelper.WriteToConsole(ConsoleColor.Black,
            $"About to simulate a sale...");

        Random random = new Random();
        string[] products = { "Apple", "Google", "Nokia", "Xiaomi", "Huawei" };

        var randomBrand = products[random.Next(products.Length)];
        var randomPrice = random.Next(1, 6) * 100; // 100, 200, 300, 400, or 500

        var nextSale = new Sale(randomPrice, randomBrand);
        _salesActor.Tell(nextSale);
    }
    
    private void HandleStop(StopSimulate message)
    {
        scheduler.Cancel();
        ConsoleHelper.WriteToConsole(ConsoleColor.DarkRed,
            "Simulation stopped");
    }
}


  • Raven에 제공하는 가이드를 유닛테스트로 옮겨두었으며 여기서 설명된 SaleActor 이벤트소싱버전을 로컬에서 수행해볼수 있습니다.

이벤트 스토어 확인

여기서 설명되는 내용은 준비된 이벤트 소싱치를 

이벤트 소싱은 저널과 스냅샷이 함께 이용되며 CRUD에서 관리되는 데이터대비 비대해질수도 있습니다.

CRUD처럼 마지막 상태데이터를 CQRS방식으로 이용하려면 DurableState Persist를 이용할수도 있습니다.

DurableState는 이벤트 발생시 변화되는  객체의 마지막 상태를 저장하는것을 의미하며 

상태로 설계된 객체가 복원을 위해 이용될수 있으며 CRUD랑 유사하게 작동하며

Journal은 발생 이벤트를 모두 기록해 이벤트 소싱과 같이 시계열이 필요하고 이벤트 재생이 필요한곳에 이용될수 있습니다.

SnapShot은 Journal과 함께 이용되어  이벤트소스로부터 너무 많은 재생이 일어나지않게 방지를 하는 장치입니다. 

샘플코드

이상 RavenDB을 Akka.net과 함께 이용해 이벤트 소싱을 구현하는 샘플을 간략하게 살펴보았으며

여기서 설명된 내용은 Akka.net기반 액터모델을 중심으로한 다양한 샘플기능들을 살펴볼수 있습니다.  




  • No labels