Ring Buffer — 고정 크기 큐로 로그와 메트릭을 처리하는 법

초기 상태:    [  |  |  |  |  |  |  |  ]
              ↑ head, tail

삽입 3개:     [ A | B | C |  |  |  |  |  ]
              ↑ head     ↑ tail

삭제 2개:     [   |   | C |  |  |  |  |  ]
                        ↑ head  ↑ tail

계속 삽입:    [   |   | C | D | E | F | G | H ]
                        ↑ head              ↑ tail

끝에 도달 후: [ I | J | C | D | E | F | G | H ]
              ↑ tail  ↑ head ← 배열 처음으로 돌아감!

public class RingBuffer<T> {
    private Object[] buffer;
    private int capacity;
    private int head;  // 읽기 위치
    private int tail;  // 쓰기 위치
    private int size;

    public RingBuffer(int capacity) {
        this.capacity = capacity;
        this.buffer = new Object[capacity];
        this.head = 0;
        this.tail = 0;
        this.size = 0;
    }

    // 삽입: O(1)
    public void offer(T item) {
        buffer[tail] = item;
        tail = (tail + 1) % capacity; // 모듈로 연산으로 순환

        if (size == capacity) {
            head = (head + 1) % capacity; // 가득 찼으면 가장 오래된 것 덮어쓰기
        } else {
            size++;
        }
    }

    // 삭제: O(1)
    @SuppressWarnings("unchecked")
    public T poll() {
        if (size == 0) return null;
        T item = (T) buffer[head];
        buffer[head] = null; // GC 도움
        head = (head + 1) % capacity;
        size--;
        return item;
    }

    // 조회: O(1)
    @SuppressWarnings("unchecked")
    public T peek() {
        if (size == 0) return null;
        return (T) buffer[head];
    }

    public boolean isEmpty() { return size == 0; }
    public boolean isFull() { return size == capacity; }
    public int size() { return size; }
}

// capacity = 1024 (2^10)
// index % 1024 == index & 1023 (비트 마스크)
int mask = capacity - 1;
tail = (tail + 1) & mask; // 모듈로보다 빠름

LinkedBlockingQueue:
- 노드 객체 생성 → GC 발생
- 락(lock)으로 동기화 → 컨텍스트 스위칭
- 약 300만 ops/sec

Disruptor:
- Ring Buffer, 객체 재사용 → GC 없음
- CAS + 메모리 배리어 → 락 프리
- 약 1억+ ops/sec

// 1. 캐시 라인 패딩 — False Sharing 방지
class Sequence {
    long p1, p2, p3, p4, p5, p6, p7; // 패딩
    volatile long value;               // 실제 값
    long p8, p9, p10, p11, p12, p13, p14; // 패딩
}

// 2. Sequence Barrier — 프로듀서/컨슈머 간 조율
// 프로듀서: "나는 여기까지 썼다" (sequence 공개)
// 컨슈머: "프로듀서가 여기까지 썼으니 여기까지 읽을 수 있다"

// 3. 다중 컨슈머 패턴
// 같은 이벤트를 여러 컨슈머가 독립적으로 처리
// 각 컨슈머가 자체 sequence를 관리

// Disruptor 사용 예시 (개념적)
// Ring Buffer 크기는 반드시 2의 거듭제곱
Disruptor<OrderEvent> disruptor = new Disruptor<>(
    OrderEvent::new,     // 이벤트 팩토리
    1024,                // Ring Buffer 크기
    DaemonThreadFactory.INSTANCE,
    ProducerType.SINGLE, // 단일 프로듀서
    new BusySpinWaitStrategy() // 대기 전략
);

// 이벤트 핸들러 체이닝
disruptor.handleEventsWith(journaler, replicator)
         .then(businessLogic)
         .then(publisher);

disruptor.start();

// 이벤트 발행
RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
long sequence = ringBuffer.next();
try {
    OrderEvent event = ringBuffer.get(sequence);
    event.setOrderId(12345);
    event.setPrice(100.0);
} finally {
    ringBuffer.publish(sequence); // 컨슈머에게 공개
}

Kafka 파티션 (로그 세그먼트):
[segment-0] [segment-1] [segment-2] [segment-3]
                                            ↑ 새 메시지 추가

오래된 세그먼트는 retention 정책에 따라 삭제
→ 논리적으로 원형 구조

// Kafka Producer — 내부적으로 RecordAccumulator가 Ring Buffer 유사 구조
Properties props = new Properties();
props.put("buffer.memory", 33554432);  // 32MB 버퍼
props.put("batch.size", 16384);        // 16KB 배치

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// RecordAccumulator가 메시지를 버퍼링한 뒤 배치로 전송

ByteBuffer buffer = ByteBuffer.allocate(1024);

// position, limit, capacity 개념
// [     쓰기 가능 영역     ]
//  ↑ position          ↑ limit      ↑ capacity

// 쓰기
buffer.put((byte) 65);  // position 이동
buffer.putInt(42);       // position 이동

// 읽기 모드 전환
buffer.flip(); // limit = position, position = 0

// 읽기
byte b = buffer.get();
int i = buffer.getInt();

// 다시 쓰기 모드
buffer.clear();   // position = 0, limit = capacity (데이터 남아있음)
buffer.compact(); // 읽지 않은 데이터를 앞으로 이동

// 서버에서 NIO로 데이터 읽기
SocketChannel channel = ...;
ByteBuffer buffer = ByteBuffer.allocateDirect(4096); // 다이렉트 버퍼

while (channel.read(buffer) > 0) {
    buffer.flip();
    processData(buffer); // 데이터 처리
    buffer.compact();    // 남은 데이터 유지
}

// 최근 N개의 로그만 메모리에 유지
public class LogBuffer {
    private final RingBuffer<LogEntry> buffer;

    public LogBuffer(int maxEntries) {
        this.buffer = new RingBuffer<>(maxEntries);
    }

    public void log(String message, Level level) {
        buffer.offer(new LogEntry(Instant.now(), message, level));
        // 가득 차면 가장 오래된 로그가 자동으로 덮어씌워짐
    }

    // 최근 로그 조회
    public List<LogEntry> getRecentLogs() {
        List<LogEntry> result = new ArrayList<>();
        // buffer를 순회하며 수집
        return result;
    }
}

// 최근 1분간의 응답 시간을 Ring Buffer로 수집
@Component
public class ResponseTimeTracker {
    private static final int WINDOW_SIZE = 60; // 60초
    private final long[] responseTimes = new long[WINDOW_SIZE];
    private int index = 0;

    public void record(long responseTimeMs) {
        responseTimes[index % WINDOW_SIZE] = responseTimeMs;
        index++;
    }

    public double averageResponseTime() {
        long sum = 0;
        int count = Math.min(index, WINDOW_SIZE);
        for (int i = 0; i < count; i++) {
            sum += responseTimes[i];
        }
        return count > 0 ? (double) sum / count : 0;
    }
}

// 스레드 안전한 Ring Buffer (ArrayBlockingQueue 사용)
BlockingQueue<Event> eventQueue = new ArrayBlockingQueue<>(1024);

// 생산자
executor.submit(() -> {
    while (true) {
        Event event = receiveEvent();
        eventQueue.put(event); // 가득 차면 대기
    }
});

// 소비자
executor.submit(() -> {
    while (true) {
        Event event = eventQueue.take(); // 비어있으면 대기
        processEvent(event);
    }
});