📸

Axon Framework Snapshot(Cache)

Date

2025/04/06

Tag

Spring

Axon

Redis

숨기기

Axon & EventSourcing

Axon Framework는 DDD, CQRS, EventSourcing을 기반으로 한 강력한 프레임워크이다.

이벤트소싱(EventSourcing)에서 어그리거트(Aggregate)가 명령(Command)을 수행할 때, 과거에 이벤트(Event)들을 순차적으로 재생하여 최신 상태를 복원한다. 하지만 이벤트 수가 많아질수록 재생 과정이 느려지고 성능 이슈로 이어질 수 있다.

이러한 문제를 해결하기 위해 스냅샷(Snapshot)이란 개념이 존재한다.

스냅샷(Snapshot)

스냅샷은 특정 시점의 어그리거트의 상태를 저장해두고, 이후에는 해당 스냅샷부터 이벤트를 재생함으로써 전체 이벤트를 처음부터 재생하는 비용을 줄여준다.

Axon에서는 두 가지의 스냅샷 생성 전략(SnapshotTriggerDefinition)을 제공한다.

AggregateLoadTimeSnapshotTriggerDefinition

어그리거트의 로딩 시간을 기반으로 스냅샷을 생성한다

EventCountSnapshotTriggerDefinition

어그리거트의 이벤트 수를 기반으로 스냅샷을 생성한다

스냅샷을 사용(JDBC 이벤트 저장소를 사용할 경우)하더라도 매 재생 시 스냅샷 정보를 DB에서 쿼리해오기 때문에 DB에 부하를 줄 수 있다. 이 또한 개선할 수 있도록 axon에서는 스냅샷 캐시 기능도 제공하고 있다.

레포지토리

EventCountSnapshotTriggerDefinition 을 사용하는 분산 환경 스냅샷 예시 코드

axon-snapshot-demo

eTaphee

사용 케이스

예제 코드에서는 3번의 이벤트마다 스냅샷이 생성되도록 설정을 했다.

@Bean
fun snapshotTrigger(snapshotter: Snapshotter): SnapshotTriggerDefinition {
    return EventCountSnapshotTriggerDefinition(snapshotter, 3)
}
Kotlin
복사

1. 단일 인스턴스 & 인메모리 캐시

단일 인스턴스 환경에서 인메모리 캐시를 사용할 경우에는 문제가 생길 수 없다. 항상 인메모리에 있는 어그리거트 버전은 최신이기 때문이다.

# 생성
FooAggregate created with id: 0

# 이벤트 저장(seq=0)
Hibernate: 
    insert 
    into
        domain_event_entry
        (aggregate_identifier, event_identifier, meta_data, payload, payload_revision, payload_type, sequence_number, time_stamp, type, global_index) 
    values
========================================================================================
# 요청 명령 수행 (스냅샷 및 이벤트를 조회하지 않음)
call increase foo value
FooAggregate value increased to: 1

# 이벤트 저장(seq=1)
Hibernate: 
    insert 
    into
        domain_event_entry
        (aggregate_identifier, event_identifier, meta_data, payload, payload_revision, payload_type, sequence_number, time_stamp, type, global_index) 
    values
        (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
========================================================================================
# 요청 명령 수행 
call increase foo value
FooAggregate value increased to: 2

# 이벤트 저장(seq=2)
Hibernate: 
    insert 
    into
        domain_event_entry
        (aggregate_identifier, event_identifier, meta_data, payload, payload_revision, payload_type, sequence_number, time_stamp, type, global_index) 
    values
        (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

# 이벤트 카운터가 3이 되었으니, 새로운 스냅샷 저장
Hibernate: 
    select
        see1_0.aggregate_identifier,
        see1_0.sequence_number,
        see1_0.type,
        see1_0.event_identifier,
        see1_0.meta_data,
        see1_0.payload,
        see1_0.payload_revision,
        see1_0.payload_type,
        see1_0.time_stamp 
    from
        snapshot_event_entry see1_0 
    where
        (
            see1_0.aggregate_identifier, see1_0.sequence_number, see1_0.type
        ) in ((?, ?, ?))
Plain Text
복사

<im.etap.domain.aggregate.FooAggregate>
  <id>0</id>
  <value>2</value>
</im.etap.domain.aggregate.FooAggregate>
XML
복사

2. 멀티 인스턴스 & 인메모리 캐시

멀티 인스턴스 환경에서는 인모메리 캐시를 사용할 경우 인스턴스 마다 캐시 저장소를 생성하기 때문에, 어그리거트 버전 불일치가 발생하여 요청이 실패하게 된다.

instance-2 가 먼저 실행되어, foo(id=0) 어그리거트를 먼저 생성 한 상태이므로, sequence(0)인 캐시가 저정된 상태이다.

첫 번째 요청(instance-1)
call increase foo value # api 요청
Hibernate: 
    select
        see1_0.type,
        see1_0.aggregate_identifier,
        see1_0.sequence_number,
        see1_0.event_identifier,
        see1_0.time_stamp,
        see1_0.payload_type,
        see1_0.payload_revision,
        see1_0.payload,
        see1_0.meta_data 
    from
        snapshot_event_entry see1_0 
    where
        see1_0.aggregate_identifier=? 
    order by
        see1_0.sequence_number desc 
    limit
        ?
Hibernate: 
    select
        dee1_0.type,
        dee1_0.aggregate_identifier,
        dee1_0.sequence_number,
        dee1_0.event_identifier,
        dee1_0.time_stamp,
        dee1_0.payload_type,
        dee1_0.payload_revision,
        dee1_0.payload,
        dee1_0.meta_data 
    from
        domain_event_entry dee1_0 
    where
        dee1_0.aggregate_identifier=? 
        and dee1_0.sequence_number>=? 
    order by
        dee1_0.sequence_number 
    limit
        ?

# instance-2에서 먼저 생성된 이벤트를 재생(sequence=0)
FooAggregate created with id: 0

# 요청으로 들어온 명령 수행
FooAggregate value increased to: 1

# 이벤트 저장(sequence=1)
Hibernate: 
    insert 
    into
        domain_event_entry
        (aggregate_identifier, event_identifier, meta_data, payload, payload_revision, payload_type, sequence_number, time_stamp, type, global_index) 
    values
        (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
Plain Text
복사

# 두 번째 요청(instance-2)
call increase foo value

# 이미 메모리에 생성에 대한 스냅샷 sequence=0이 남아 있으므로, 생성 이벤트는 재생하지 않는다.
FooAggregate value increased to: 1

# 이벤트 저장(sequence=1)
Hibernate: 
    insert 
    into
        domain_event_entry
        (aggregate_identifier, event_identifier, meta_data, payload, payload_revision, payload_type, sequence_number, time_stamp, type, global_index) 
    values
        (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

# instance-1에서 sequence=1에 대한 이벤트가 저장되어 있으므로 중복키 위반이 발생한다.
Error: 1062-23000: Duplicate entry '0-1' for key 'UK8s1f994p4la2ipb13me2xqm1w'
SQL Error: 1062, SQLState: 23000
(conn=55) Duplicate entry '0-1' for key 'UK8s1f994p4la2ipb13me2xqm1w'
Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed: org.axonframework.eventsourcing.eventstore.EventStoreException: An event for aggregate [0] at sequence [1] was already inserted] with root cause
Plain Text
복사

3. 멀티 인스턴스 & 분산 캐시

어그리거트 명령 수행 시 캐시 스토어에 있는 최신 어그리거트를 가져와서 수행하기 때문에 버전 불일치 문제가 생기지 않는다.

# instance-2 시작
FooAggregate created with id: 0

# FooCreatedEvent
Hibernate: 
    insert 
    into
        domain_event_entry
        (aggregate_identifier, event_identifier, meta_data, payload, payload_revision, payload_type, sequence_number, time_stamp, type, global_index) 
    values
        (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
Plain Text
복사

# instance-1 요청
call increase foo value

# 캐시에 어그리거트 정보가 없어 이벤트 스토어에서 스냅샷&이벤트 정보를 조회
Hibernate: 
    select
        see1_0.type,
        see1_0.aggregate_identifier,
        see1_0.sequence_number,
        see1_0.event_identifier,
        see1_0.time_stamp,
        see1_0.payload_type,
        see1_0.payload_revision,
        see1_0.payload,
        see1_0.meta_data 
    from
        snapshot_event_entry see1_0 
    where
        see1_0.aggregate_identifier=? 
    order by
        see1_0.sequence_number desc 
    limit
        ?
        
Hibernate: 
    select
        dee1_0.type,
        dee1_0.aggregate_identifier,
        dee1_0.sequence_number,
        dee1_0.event_identifier,
        dee1_0.time_stamp,
        dee1_0.payload_type,
        dee1_0.payload_revision,
        dee1_0.payload,
        dee1_0.meta_data 
    from
        domain_event_entry dee1_0 
    where
        dee1_0.aggregate_identifier=? 
        and dee1_0.sequence_number>=? 
    order by
        dee1_0.sequence_number 
    limit
        ?

====================================================================================

# 조회한 이벤트를 재생 & 명령 수행
FooAggregate created with id: 0
FooAggregate value increased to: 1
Hibernate: 
    insert 
    into
        domain_event_entry
        (aggregate_identifier, event_identifier, meta_data, payload, payload_revision, payload_type, sequence_number, time_stamp, type, global_index) 
    values
        (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
Plain Text
복사

# instance-2 요청
call increase foo value

# 캐시에서 어그리거트 조회 후 명령 수행
FooAggregate value increased to: 2

Hibernate: 
    insert 
    into
        domain_event_entry
        (aggregate_identifier, event_identifier, meta_data, payload, payload_revision, payload_type, sequence_number, time_stamp, type, global_index) 
    values
        (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

# 이벤트 카운터가 3, 이전 스냅샷 존재 여부 확인    
Hibernate: 
    select
        see1_0.type,
        see1_0.aggregate_identifier,
        see1_0.sequence_number,
        see1_0.event_identifier,
        see1_0.time_stamp,
        see1_0.payload_type,
        see1_0.payload_revision,
        see1_0.payload,
        see1_0.meta_data 
    from
        snapshot_event_entry see1_0 
    where
        see1_0.aggregate_identifier=? 
    order by
        see1_0.sequence_number desc 
    limit
        ?
        
Hibernate: 
    select
        dee1_0.type,
        dee1_0.aggregate_identifier,
        dee1_0.sequence_number,
        dee1_0.event_identifier,
        dee1_0.time_stamp,
        dee1_0.payload_type,
        dee1_0.payload_revision,
        dee1_0.payload,
        dee1_0.meta_data 
    from
        domain_event_entry dee1_0 
    where
        dee1_0.aggregate_identifier=? 
        and dee1_0.sequence_number>=? 
    order by
        dee1_0.sequence_number 
    limit
        ?

# 새로운 스냅샷 생성
FooAggregate created with id: 0
FooAggregate value increased to: 1
FooAggregate value increased to: 2
Hibernate: 
    select
        see1_0.aggregate_identifier,
        see1_0.sequence_number,
        see1_0.type,
        see1_0.event_identifier,
        see1_0.meta_data,
        see1_0.payload,
        see1_0.payload_revision,
        see1_0.payload_type,
        see1_0.time_stamp 
    from
        snapshot_event_entry see1_0 
    where
        (
            see1_0.aggregate_identifier, see1_0.sequence_number, see1_0.type
        ) in ((?, ?, ?))
Hibernate: 
    insert 
    into
        snapshot_event_entry
        (event_identifier, meta_data, payload, payload_revision, payload_type, time_stamp, aggregate_identifier, sequence_number, type) 
    values
        (?, ?, ?, ?, ?, ?, ?, ?, ?)
Hibernate: 
    delete see1_0 
    from
        snapshot_event_entry see1_0 
    where
        see1_0.aggregate_identifier=? 
        and see1_0.sequence_number<?
Plain Text
복사

JCacheAdapter

어그리거트의 캐시를 redis에 저장하기 위해 JCacheAdapter 를 사용했다. JCache는 Java 표준 캐싱 API로, javax.cache 패키지에 포함되어 있으며, 구현체로는 Redisson을 사용했다.

JCacheAdatper 사용 시, RedissonClient를 등록해야 하며, 이 때 코덱을 추가로 지정해야 한다. Axon은 기본적으로 XML 포맷 기반의 XStreamSerializer 를 사용하므로, 이에 맞는 코덱을 별도로 구현해야한다.

@Profile("redis")
@Bean(name = ["snapshotCache"])
fun redisCache(@Qualifier("axonCacheManager") cacheManager: CacheManager): Cache {
    val config = MutableConfiguration<String, Any>()
    config.setExpiryPolicyFactory(CreatedExpiryPolicy.factoryOf(Duration.ONE_DAY))
    if (!cacheManager.cacheNames.contains(SNAPSHOT_CACHE_NAME)) {
        cacheManager.createCache(SNAPSHOT_CACHE_NAME, config)
    }
    return JCacheAdapter(cacheManager.getCache(SNAPSHOT_CACHE_NAME))
}

@Bean(name = ["axonRedissonClient"])
fun axonCacheRedissonClient(serializer: Serializer, redisProperties: RedisProperties): RedissonClient {
    return Redisson.create(Config().apply {
        useSingleServer().apply {
            address = buildRedisUri(redisProperties)
            username = redisProperties.username
            password = redisProperties.password
        }
        codec = AxonCacheCodec(serializer)
    })
}

@Bean(name = ["axonCacheManager"])
fun axonCacheManager(
    @Qualifier("axonRedissonClient") redissonClient: RedissonClient,
    redisProperties: RedisProperties
): CacheManager {
    return JCacheManager(
        redissonClient as Redisson,
        javaClass.classLoader,
        Caching.getCachingProvider(),
        Properties(),
        URI.create(buildRedisUri(redisProperties))
    )
}

class AxonCacheCodec(
    private val serializer: Serializer,
) : BaseCodec() {

    private val encoder = Encoder { obj ->
        val buf: ByteBuf = ByteBufAllocator.DEFAULT.buffer()
        try {
            ByteBufOutputStream(buf).use { os ->
                val serialized = serializer.serialize(obj, ByteArray::class.java)
                val data = serialized.data as ByteArray
                os.write(data)
                os.flush()
            }
            buf
        } catch (e: IOException) {
            buf.release()
            throw e
        } catch (e: Exception) {
            buf.release()
            throw IOException(e)
        }
    }

    private val decoder = Decoder { buf, _ ->
        ByteBufInputStream(buf).use { input ->
            val bytes = ByteArray(buf.readableBytes())
            input.read(bytes)

            val xmlString = String(bytes, StandardCharsets.UTF_8)
            val simpleSerializedObject = SimpleSerializedObject(
                xmlString,
                String::class.java,
                AggregateCacheEntry::class.qualifiedName,
                ""
            )

            return@Decoder serializer.deserialize<String, Any>(simpleSerializedObject)
        }
    }

    override fun getValueDecoder(): Decoder<Any> {
        return decoder
    }

    override fun getValueEncoder(): Encoder {
        return encoder
    }
}
Kotlin
복사