Pub/Sub Pattern

정의

• Observer 패턴의 일종
• Pub가 Broker를 통해 이벤트를 전달하고, 다수의 Sub는 자신이 구독한 topic의 이벤트를 비동기적으로 수신 및 처리하도록 해주는 패턴

Observer 패턴

https://monktech.tistory.com/167

Observer Pattern

• 위는 Observer 패턴의 동작 흐름

 

 

Observer 와 Pub-Sub의 차이점

• Topic/Event chanel 의 존재 유무
• Pub가 특정 Topic에 메시지 발생하면, Broker가 해당 Topic을 구독한 Sub에게 메시지를 전달
  • 그렇다보니, Pub는 Sub가 누구인지 알 필요없다
  • Observer는 객체 참조 수준의 결합으로 Sub가 누군인지는 알아야한다.
• Observer가 객체 참조를 통한 직접적인 이벤트 통지에 초점을 맞춘다면, Pub/Sub은 발행자와 구독자의 독립성을 높이는 메시지 전달 구조에 초점을 맞춤

 

구성 요소 및 역할

Publisher (발행자)	메시지 또는 이벤트를 생성하고 발행
Subscriber (구독자)	관심 있는 메시지를 구독하고 처리
Broker (중개자)	메시지를 받아 적절한 구독자에게 전달

• Topic은 메시지를 분류하는 논리적인 채널

 

동작 원리

예를 들어 쇼핑몰에서 사용자가 주문을 생성했다고 가정

• 동작 순서
  • 사용자 주문 생성 > 주문 서비스(Publisher)가 order.created 이벤트를 발행
  • Broker가 이벤트를 수신
  • Broker가 order.created Topic을 구독 중인 서비스에 이벤트를 전달
  • 재고, 알림, 배송 서비스는 전달된 이벤트를 각각 독립적으로 처리

여기서 중요한 점은 주문 서비스가 재고 알림 배송 서비스를 직접 호출하지 않아도 된다는 것

 

 

Pub/Sub를 사용하는 이유

기존의 직접 호출 방식에서는 주문 서비스가 여러 서비스를 호출

 

주문 서비스
 ├── 재고 서비스 호출
 ├── 알림 서비스 호출
 └── 배송 서비스 호출

• 서비스가 추가될 때마다 주문 서비스의 코드가 변경될 가능성 상승

주문 서비스
     │
     ▼
  이벤트 발행
     │
     ▼
    Topic
     │
     ├── 재고 서비스
     ├── 알림 서비스
     └── 배송 서비스

• 이벤트만 발행하고 나머지 작업은 구독 서비스가 수행하므로, 서비스 간 의존성을 낮춤

그 외, 장점으로는 새운 Sub를 쉽게 추가, Pub가 Sub의 작업완료를 기다리지 않아도 된다는 비동기 처리, 각 서비스가 독립적으로 이벤트 처리

 

그러나 단점으로는

• 브로커의 저장 전달 정책에 따라 메시지 유실 가능성
• 이벤트가 여러 서비스로 전파 되므로 추적 복잡
• 이벤트 처리 순서가 항상 보장되지는 않음
• 메시지가 두 번 이상 전달될 가능성에 대비도 해야함

Pub/Sub와 Message Queue 차이

Pub/Sub	Message Queue
하나의 이벤트를 여러 구독자에게 전달 가능	일반적으로 하나의 작업을 소비자 중 하나가 처리
이벤트 전파 중심	작업 분산 중심
예: 주문 생성 이벤트 알림	예: 이미지 처리 작업 분배

 

실제로는 두 모델이 함께 사용되기도 한다. Kafka의 Consumer Group이나 RabbitMQ의 Exchange·Queue 구조가 대표적인 예이다.

 

주된 사용처

• 독립된 여러 서비스 간에 통신할 때
  • 용자가 "회원가입" 버튼을 누르면, User-Service는 DB에 저장하고 끝나는 게 아니라 "UserSignedUp" 이라는 이벤트를 브로커에 전
  • Notification-Service는 이 이벤트를 구독(Subscribe)하고 있다가 환영 이메일
  • Point-Service는 이 이벤트를 구독하고 있다가 가입 축하 포인트를 지급
• 실시간 데이터 스트리밍 & 대용량 로그 수집 (Big Data / Logging)
  • 전 세계 사용자의 클릭 스트림, 서버 에러 로그, IoT 센서 데이터를 실시간으로 모아서 처리
  • 대용령 로그를 Kafka 같은 거대한 Pub/Sub 브로커에 넣면, 분석 서버, 모니터링 서버(Elasticsearch 등), 백업 서버들이 각각 필요한 만큼 가져가서 처리
• 비동기 백그라운드 작업 (Task Queues / Job Processing)
  • 사용자가 웹사이트에서 무거운 작업을 요청했을 때, 화면이 멈추지 않게 뒤로 밀어두고 처리
  • 유튜브에 고화질 영상을 업로드하면, 업로드 서버는 일단 "완료" 응답을 주고, 내부 Pub/Sub 큐에 작업
    • 백그라운드의 워커들이 인코딩, 썸네일 추출 등의 작업을 알아서 나눠서 처리
• 실시간 알림 및 채팅 시스템 (Real-time Messaging)
  • 웹소켓(WebSocket) 기반의 채팅방이나 실시간 푸시 알림 서버 구조
  • 카카오톡이나 슬랙 같은 곳에서 특정 채팅방(room-123)에 메시지를 보내면, 그 방을 구독하고 있는 모든 사용자(클라이언트)들에게 실시간으로 메시지가 브로드캐스팅

 

구현

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.BlockingQueue
import java.util.concurrent.LinkedBlockingQueue
import java.util.Scanner
import kotlin.concurrent.thread


class Publisher(private val broker: MessageBroker) {

    fun publishNumber(number: Int) {
        val mod = number % 5
        val topic = "topic-mod-$mod"
        val message = "Number-$number"

        broker.publish(topic, message)
        println(" [Publisher] -> [$topic]으로 라우팅 완료: $message")
    }
}


class MessageBroker(partitionCount: Int = 5) {
    private val topicQueues = ConcurrentHashMap<String, BlockingQueue<String>>()

    init {
        println("=== [MessageBroker] 토픽 큐 자동 초기화 시작 ===")
        for (i in 0 until partitionCount) {
            val topic = "topic-mod-$i"
            topicQueues[topic] = LinkedBlockingQueue()
            println(" [MessageBroker] 토픽 큐 생성 완료: $topic")
        }
        println("==========================================")
    }

    fun publish(topic: String, message: String) {
        val queue = topicQueues[topic] ?: throw IllegalStateException("존재하지 않는 토픽입니다: $topic")
        queue.put(message)
    }

    fun fetchMessage(topic: String): String {
        val queue = topicQueues[topic] ?: throw IllegalStateException("존재하지 않는 토픽입니다: $topic")
        return queue.take()
    }

    fun getRemainingMessageCounts(topics: List<String>): Map<String, Int> {
        val remainingCounts = mutableMapOf<String, Int>()
        for (topic in topics) {
            val queue = topicQueues[topic]
            remainingCounts[topic] = queue?.size ?: 0
        }
        return remainingCounts
    }
}

class Subscriber(
    private val id: Int,
    private val broker: MessageBroker,
    private val targetTopic: String,
    private val counts: IntArray
) : Runnable {

    override fun run() {
        println("Subscriber-$id 시작됨 (담당 주제: $targetTopic)")

        while (!Thread.currentThread().isInterrupted) {
            try {
                val message = broker.fetchMessage(targetTopic)
                counts[id]++
                println("   [Subscriber-$id] 수신 성공 ([$targetTopic]): $message")
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                break
            } catch (e: Exception) {
                println("   [Subscriber-$id] 처리 중 오류 발생: ${e.message}")
            }
        }

        println("Subscriber-$id 종료됨.")
    }
}

fun main() {
    val topics = (0..4).map { "topic-mod-$it" }
    val broker = MessageBroker(partitionCount = 5)
    val publisher = Publisher(broker)
    val scanner = Scanner(System.`in`)
    val processedCounts = IntArray(10)

    println("=== Pub/Sub 시스템 시작 ===")

    val threads = mutableListOf<Thread>()
    for (id in 0..9) {
        val targetTopic = topics[id % topics.size]
        val subscriber = Subscriber(
            id = id,
            broker = broker,
            targetTopic = targetTopic,
            counts = processedCounts
        )
        val t = thread(start = true) {
            subscriber.run()
        }
        threads.add(t)
    }

    println("10개의 Subscriber(0~9)가 5개의 주제에 분산 배치되었습니다.")
    println("0을 입력하면 프로그램이 종료됩니다.\n")

    while (true) {
        print("발행 방식을 선택하세요:\n1) 1부터 N까지 연속 자동 발행\n2) 직접 숫자 1개 입력\n선택 (종료: 0): ")
        try {
            val choice = scanner.nextInt()

            if (choice == 0) {
                println("프로그램을 종료하는 중...")
                threads.forEach { it.interrupt() }
                threads.forEach { it.join() }
                println("프로그램이 정상 종료되었습니다.")
                break
            } else if (choice == 1) {
                processedCounts.fill(0)

                print("몇까지 발행할까요? (예: 100, 1000): ")

                val maxNumber = scanner.nextInt()

                println("--- 1부터 $maxNumber 까지 숫자 발행 시작 ---")
                for (number in 1..maxNumber) {
                    publisher.publishNumber(number)
                }

                println("--- 발행 완료 ---")

                Thread.sleep(600)

                println("\n📊 [구독자 스레드별 처리량 통계]")
                for (id in 0..9) {
                    val count = processedCounts[id]
                    val assignedTopic = topics[id % topics.size]
                    println(" - Subscriber-$id (담당: $assignedTopic): 총 $count 건 처리")
                }

                println("\n📊 [큐 잔여 데이터 현황 검사]")
                val sizes = broker.getRemainingMessageCounts(topics)
                for (topic in topics) {
                    val remaining = sizes[topic] ?: 0
                    println(" - $topic 잔여 개수: $remaining 개")
                }
                println("----------------------------------------\n")

            } else if (choice == 2) {
                print("발행할 숫자를 입력하세요: ")
                val number = scanner.nextInt()
                publisher.publishNumber(number)
            } else {
                println("올바른 번호를 입력해주세요.")
            }
        } catch (e: Exception) {
            println("올바른 숫자를 입력해주세요 또는 에러 발생: ${e.message}")
            scanner.next()
        }
    }
}

 

 

Publisher

class Publisher(private val broker: MessageBroker) {

    fun publishNumber(number: Int) {
        val mod = number % 5
        val topic = "topic-mod-$mod"
        val message = "Number-$number"

        broker.publish(topic, message)
        println(" [Publisher] -> [$topic]으로 라우팅 완료: $message")
    }
}

• 주제별 라우팅 역할
• 본 코드에서 선정한 주제: 특정 정수를 5로 나누었을 때 나머지 값
  • 나머지 값에 따라 서로 다른 큐에 배치
  • 주제는 나머지 값, 메시지는 원본 값
• 메시지 브로커 안에 해시맵 자료구조 안에 BlockingQueue 가 여러 개 있는데 주제에 맞는 큐 안에 메시지를 전달하도록하는 라우팅하는 역할
  • 라우팅 원리:정수값 원본 메시지를 전달 받으면 5로 나누었을 때 나머지 값으로 하는 주제의 큐에 원본 메시지를 브로커의 주제에 맞는 큐에 전달
• 브로커는 전달 받으면 가공 메시지 값을 각 주제의 큐에 적재

 

MessageBroker

class MessageBroker(partitionCount: Int = 5) {
    private val topicQueues = ConcurrentHashMap<String, BlockingQueue<String>>()

    init {
        println("=== [MessageBroker] 토픽 큐 자동 초기화 시작 ===")
        for (i in 0 until partitionCount) {
            val topic = "topic-mod-$i"
            topicQueues[topic] = LinkedBlockingQueue()
            println(" [MessageBroker] 토픽 큐 생성 완료: $topic")
        }
        println("==========================================")
    }

    fun publish(topic: String, message: String) {
        val queue = topicQueues[topic] ?: throw IllegalStateException("존재하지 않는 토픽입니다: $topic")
        queue.put(message)
    }

    fun fetchMessage(topic: String): String {
        val queue = topicQueues[topic] ?: throw IllegalStateException("존재하지 않는 토픽입니다: $topic")
        return queue.take()
    }

    fun getRemainingMessageCounts(topics: List<String>): Map<String, Int> {
        val remainingCounts = mutableMapOf<String, Int>()
        for (topic in topics) {
            val queue = topicQueues[topic]
            remainingCounts[topic] = queue?.size ?: 0
        }
        return remainingCounts
    }
}

• 생성자 호출을 통한 초기화 세팅은 init 코드 안에서 해결
  • main 함수에서 partitionCount의 파라미터로 5 라는 값이 전달
  • 그래서 총 5개의 주제별 큐가 생성
  •  [MessageBroker] 토픽 큐 생성 완료: topic-mod-0
     [MessageBroker] 토픽 큐 생성 완료: topic-mod-1
     [MessageBroker] 토픽 큐 생성 완료: topic-mod-2
     [MessageBroker] 토픽 큐 생성 완료: topic-mod-3
     [MessageBroker] 토픽 큐 생성 완료: topic-mod-4
  • 그래서 만약 1000 이라는 값이 메시지로 들어오면 5로 나누었을 때 0이므로 0 이라는 토픽큐에 전달됨
• ConcurrentHashMa 을 쓴 이유
  • 읽기 성능을 극대화하기 위해
  • 읽기 작업에 있어서는 락을 안건다
  • 서로 다른 버킷에 대해서는 읽기 쓰기 관련 경합성을 따지지 않음
    • 그래서 A 버킷에 대해 누가 쓰고 있다면, B 버킷을 읽을려고하는 다른 스레드는 다른 버킷 A에서 뭘 하든 신경 안쓰고 읽고 쓴다
  • 읽는 작업에 대해서는 애초에 락을 걸지 않아서 여러 스레드 접근 가능 
  • 특히나 쓰기 작업을 하더라고 특정 블록의 첫번째 노드에 syncronized 를 걸게되며, 값 자체와 포인터는 volatile이 적용된다. 그래서 캐시 단위에서 메모리 단위에서 수행되는 전체적으로 갱신.
    • 이때, JVM 자체가 연산 중간의 어떤 과정에서의 값을 읽기 작업에서 읽는 것을 방지하기에, 읽기 작업시에는 아예 변경되기 전의 값을 읽거나, 변경된 후의 값을 받아들이게됨
• BlockingQueue를 쓴 이유
  • 큐에 데이터 없으면 스레드를 Blocking 처리해서 CPU 계속 쓰게 하는 걸 방지할려고
  • 큐가 비어 있는 상태에서 소비자가 take()를 호출하면, 큐는 내부 조건 변수를 통해 park() 또는 운영체제 커널의 대기 큐(Wait Queue) 등록 시스템 콜을 호출
    • park(): 말그대로 파킹하는 거 생각하면 편함. 여기서는 heap 메모리에 TCB 값들 파킹해둠
  • 이 스레드를 CPU 실행 큐(Run Queue)에서 제외하고 배정된 CPU 코어에서 해제
  • 새로운 데이터가 put()되어 큐에 적재한 직후, notEmpty.signal()(내부적으로 unpark 또는 커널의 wake 시스템 콜)을 호출하면 운영체제 스케줄러는 대기 큐에 있던 소비자 스레드를 다시 실행 큐(RUNNABLE)로 이동시키며, 다음 스케줄링 시점에 CPU 코어를 할당하여 작업을 재개
• publish는 방금 Publisher가 각 주제별 큐에 데이터 넣는 거, fetchMessage는 Subscriber 즉 구독자가 데이터 가져가는거, getRemainingMessageCounts는 나중에 통계 낼 때

 

Subscriber

class Subscriber(
    private val id: Int,
    private val broker: MessageBroker,
    private val targetTopic: String,
    private val counts: IntArray
) : Runnable {

    override fun run() {
        println("Subscriber-$id 시작됨 (담당 주제: $targetTopic)")

        while (!Thread.currentThread().isInterrupted) {
            try {
                val message = broker.fetchMessage(targetTopic)
                counts[id]++
                println("   [Subscriber-$id] 수신 성공 ([$targetTopic]): $message")
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
                break
            } catch (e: Exception) {
                println("   [Subscriber-$id] 처리 중 오류 발생: ${e.message}")
            }
        }

        println("Subscriber-$id 종료됨.")
    }
}

• !Thread.currentThread().isInterrupted: 시스템 종료시 while 루프에서 빠져나갈 수 있게 
• val message = broker.fetchMessage(targetTopic): 각 스레드가 run을 통해 각자의 작업 흐름을 실행시킬 텐데, 각자의 주제별 큐에 맞게 자신들의 값을 가져올 수 있게 targetTopic을 파라미터로서 전달하여 메시지 큐에 값을 가져옴
•     catch (e: InterruptedException) {
                  Thread.currentThread().interrupt()
                  break
     } catch (e: Exception) {
                  println("   [Subscriber-$id] 처리 중 오류 발생: ${e.message}")
              }
  • 첫번째 error 캐치는 main 함수에서 종료 요청 보낼 때의 인터럽트를 받기위해
  • 두번쨰 error 캐치는 주제별 큐에서 메시지 가져올때 가져오는 걸 실패한 걸 알리기위한 것

 

main 함수

fun main() {
    val topics = (0..4).map { "topic-mod-$it" }
    val broker = MessageBroker(partitionCount = 5)
    val publisher = Publisher(broker)
    val scanner = Scanner(System.`in`)
    val processedCounts = IntArray(10)

    println("=== Pub/Sub 시스템 시작 ===")

    val threads = mutableListOf<Thread>()
    for (id in 0..9) {
        val targetTopic = topics[id % topics.size]
        val subscriber = Subscriber(
            id = id,
            broker = broker,
            targetTopic = targetTopic,
            counts = processedCounts
        )
        val t = thread(start = true) {
            subscriber.run()
        }
        threads.add(t)
    }

    println("10개의 Subscriber(0~9)가 5개의 주제에 분산 배치되었습니다.")
    println("0을 입력하면 프로그램이 종료됩니다.\n")

    while (true) {
        print("발행 방식을 선택하세요:\n1) 1부터 N까지 연속 자동 발행\n2) 직접 숫자 1개 입력\n선택 (종료: 0): ")
        try {
            val choice = scanner.nextInt()

            if (choice == 0) {
                println("프로그램을 종료하는 중...")
                threads.forEach { it.interrupt() }
                threads.forEach { it.join() }
                println("프로그램이 정상 종료되었습니다.")
                break
            } else if (choice == 1) {
                processedCounts.fill(0)

                print("몇까지 발행할까요? (예: 100, 1000): ")

                val maxNumber = scanner.nextInt()

                println("--- 1부터 $maxNumber 까지 숫자 발행 시작 ---")
                for (number in 1..maxNumber) {
                    publisher.publishNumber(number)
                }

                println("--- 발행 완료 ---")

                Thread.sleep(600)

                println("\n📊 [구독자 스레드별 처리량 통계]")
                for (id in 0..9) {
                    val count = processedCounts[id]
                    val assignedTopic = topics[id % topics.size]
                    println(" - Subscriber-$id (담당: $assignedTopic): 총 $count 건 처리")
                }

                println("\n📊 [큐 잔여 데이터 현황 검사]")
                val sizes = broker.getRemainingMessageCounts(topics)
                for (topic in topics) {
                    val remaining = sizes[topic] ?: 0
                    println(" - $topic 잔여 개수: $remaining 개")
                }
                println("----------------------------------------\n")

            } else if (choice == 2) {
                print("발행할 숫자를 입력하세요: ")
                val number = scanner.nextInt()
                publisher.publishNumber(number)
            } else {
                println("올바른 번호를 입력해주세요.")
            }
        } catch (e: Exception) {
            println("올바른 숫자를 입력해주세요 또는 에러 발생: ${e.message}")
            scanner.next()
        }
    }
}

val threads = mutableListOf<Thread>()
for (id in 0..9) {
    val targetTopic = topics[id % topics.size]
    val subscriber = Subscriber(
        id = id,
        broker = broker,
        targetTopic = targetTopic,
        counts = processedCounts
    )
    val t = thread(start = true) {
        subscriber.run()
    }
    threads.add(t)
}

• 플랫폼 스레드 10개 생성하고 그 객체를 threads 라는 배열안에 넣는 코드
• 0~9 번 스레드는 0,1,2,3,4 라는 주제별 토픽에 매핑 시키기위해 

targetTopic = topics[id % topics.size]

• 라는 코드 사용
• start = true
  • Thread.start()를 호출
  • JVM은 운영체제 커널에 시스템 콜을 보내 실제 OS 스레드를 생성하고, 해당 스레드의 콜 스택을 할당한 뒤, OS 스케줄러의 실행 큐에 등록하여 CPU 자원을 할당받을 수 있는 RUNNABLE 상태로
  • start = false로 설정하면 메모리에 자바 Thread 객체만 생성될 뿐 OS 스레드는 생성되지 않음
• subscriber.run()
  • thread { ... } 람다 블록 내부에 있기 때문에, 메인 스레드가 아니라 새로 생성된 저 하부 OS 스레드가 CPU를 할당받아 실행을 시작할 때 최초로 진입