[Spring/batch] BlockingQueue와 스레드를 활용한 Telegram 429 Rate Limit 해결

문제 상황

 스프링 배치 서비스에서 여러 작업 대상을 스레드 풀로 병렬 처리하고, 각 스레드가 완료 후 Telegram으로 배치 결과를 알림하는 기능이 있다.

 

 평소에는 문제가 없었는데, 단기간(1~2초)에 끝나는 배치 대상이 많은 특정 시간대의 배치에서만 아래와 같은 에러가 반복 발생했다.

 

 

 14:30:00에 46건을 대상으로 배치 시작 후, 순차적으로 Telegram 알림을 발송했다.(비즈니스 로직 수행부터 알림 발송까지 배치 대상 당 1~2초 안에 끝난다.)

 그러다가 14:30:33부터 Telegram 알림 발송 에러가 발생했다.

 

14:30:33  429 Too Many Requests  retry_after: 27
14:30:34  429 Too Many Requests  retry_after: 26
14:30:35  429 Too Many Requests  retry_after: 25

 

429 Too Many Requests가 반환되며, 응답 본문에 retry_after 값(초)이 포함된다.

 

 

 

 

상세 로직

 

 

 여러 작업을 동시에 처리하기 위해 스레드 풀을 사용한다. 스레드 풀 안에는 worker-1, worker-2.. 다수의 워커 스레드가 존재하고, 각 스레드는 독립적으로 배치 작업을 처리한다.

 

 작업이 끝나면 결과에 따라 알림 경로가 나뉜다. 성공했을 때는 sendSuccess()를 통해 채팅방 A로, 실패했을 때는 sendFailure()를 통해 채팅방 B로 Telegram 알림이 전송된다. 두 워커가 병렬로 실행되기 때문에 이 알림들은 거의 동시에 발송된다.

 

 배치 작업 대상이 많아질수록 두 워커는 빠르게 작업을 완료하고 연속적으로 알림을 쏟아낸다. 각 채팅방이 받는 메시지 건수가 쌓이다 보면 어느 순간 Telegram의 전송량 제한을 초과하게 되고, 그때부터 429 Too Many Requests 에러가 반복해서 발생한다.

 

 

 

---

원인 분석

Telegram Bot Rate Limit 정책

 Telegram Bot API는 과도한 메시지 전송을 막기 위해 Rate Limit 정책을 두고 있다.

텔레그램 봇 공식 문서 - https://core.telegram.org/bots/faq#my-bot-is-hitting-limits-how-do-i-avoid-this

 

상황	제한
같은 채팅방	초당 1건 권장
그룹 채팅	분당 최대 20건
전체 (모든 채팅방 합산)	초당 약 30건

 

 

 이 로직에서 핵심이 되는 규칙은 같은 채팅방에 분당 최대 20건이다. 채팅방 전체 합산이 아니라 채팅방 하나를 기준으로 카운팅한다.

 

 채팅방이 A, B로 분리되어 있으니 괜찮지 않을까 싶지만, 실제로는 그렇지 않았다. 이 배치 서비스에서 같은 채팅방에 메시지가 쌓이는 경로가 두 가지 있다.

 

 1.같은 채팅방으로 향하는 다른 배치 작업이 동시에 실행되는 경우

 채팅방 A가 성공 알림방이라면, 이 서비스뿐 아니라 다른 배치 타입의 성공 알림도 같은 채팅방 A로 전송될 수 있다. 배치가 여러 개 겹쳐 실행되는 시간대에는 채팅방 A로 향하는 메시지가 여러 배치에서 동시에 몰려들고, 합산 건수가 20건을 넘기게 된다.

 

 2. 배치 작업 대상 자체가 많은 경우

 작업 대상이 수십 개라면 두 워커는 작업을 마칠 때마다 계속 알림을 전송한다. 워커가 빠르게 작업을 완료할수록 알림도 빠르게 쌓이고, 60초 안에 같은 채팅방으로 20건 이상이 누적되면 그 순간 429가 발생한다.

 

 

 

 

 

슬라이딩 윈도우

 Telegram은 슬라이딩 윈도우 방식으로 제한을 적용한다.

 

 

 배치가 시작되면 병렬 워커들이 작업을 마칠 때마다 Telegram 알림을 연속으로 전송한다. Telegram은 이 전송량을 슬라이딩 윈도우 방식으로 측정하는데, '지금 이 순간부터 과거 60초' 안에 들어온 메시지를 실시간으로 카운팅한다.

 

 14:30:00에 배치가 시작된 후 33초 만에 메시지가 20건을 넘긴다. 그 순간 429가 반환되고, 응답에 담긴 retry_after 값이 매초 1씩 줄어드는 이유도 여기에 있다. 고정 윈도우였다면 14:31:00이 되는 순간 카운터가 리셋되겠지만, 슬라이딩 윈도우는 오래된 메시지가 60초 밖으로 밀려나야 비로소 한 건씩 여유가 생기기 때문이다. 14:31:00이 되어서야 retry_after가 0이 되고 전송이 재개된다.

 

 

 

 

---

해결책 후보

세 가지 방법을 검토했다.

방식	동시 발송 방지	워커 스레드 블록
sleep만 추가	X	X
synchronized + sleep	O	O
단일 BlockingQueue + 전용 스레드	O	X (비동기)

 

 

 

1. Thread.sleep() 추가

 

 처음 떠오른 해결책은 메서드에 Thread.sleep()을 추가하는 것이었다.

public void sendSuccess(...) {
    Thread.sleep(1000); // 각 스레드가 1초 잠든 뒤
    telegramProcess.send(message); // 동시에 여기 도달 → 여전히 동시 발송
}

 

 만약 10개 스레드가 동시에 sleep(1000)을 호출하면, 1초 후에 동시에 깨어나 동시에 전송한다. sleep은 각 스레드의 실행을 지연시킬 뿐, 전송 순서를 직렬화하지 않는다.

 

 

 

2. synchronized + sleep

@Service
public class TelegramNotificationService {
    private final Object lock = new Object();
    private long lastSentMs = 0;

    public void send(String chatId, String token, String message) {
        synchronized (lock) {
            long elapsed = System.currentTimeMillis() - lastSentMs;
            if (elapsed < 3000) {
                Thread.sleep(3000 - elapsed);
            }
            TelegramApi.send(chatId, token, message);
            lastSentMs = System.currentTimeMillis();
        }
    }
}

 

 synchronized로 진입을 직렬화하고, 마지막 전송 이후 3초가 지나지 않았으면 나머지 시간만큼 sleep한다.

 

 동시 발송은 막지만 워커 스레드가 synchronized  블록 안에서 발송 완료까지 블록된다. 처리 대상이 많을수록 마지막 스레드의 대기 시간이 길어진다.

 

 

 

 

3. BlockingQueue + 전용 스레드

@Service
public class TelegramNotificationService {
    private final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(500);

    @PostConstruct
    public void startSender() {
        Thread.ofVirtual().name("telegram-sender").start(() -> {
            while (true) {
                try {
                    Runnable task = queue.take();
                    task.run();
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
    }

    public void send(String chatId, String token, String message) {
        queue.offer(() -> TelegramApi.send(chatId, token, message));
    }
}

 

 워커 스레드는 queue.offer()만 호출하고 즉시 리턴한다. 실제 전송은 전용 스레드 하나가 3초 간격으로 처리한다. 전체 전송량이 최대 20건/분(60초 ÷ 3초)이 되므로, 어느 채팅방으로 보내든 Rate Limit에 걸리지 않는다.

 

 

 

 

단일 큐의 한계 - 채팅방별 독립 큐로

 단일 큐 방식은 Rate Limit 문제를 해결하지만, 채팅방이 여러 개일 때 한 가지 고민이 생긴다. (실제로 채팅방이 2개뿐 아니라 더 있다.)

 

 단일 큐에서는 채팅방 A와 채팅방 B의 메시지가 같은 줄을 서게 된다. 채팅방 A에 메시지가 몰리면, 채팅방 B의 메시지는 채팅방 A의 메시지가 빠져나갈 때까지 대기해야 한다. 전체 처리량도 전 채팅방 합산 20건/분에 묶인다.

 

비교 항목	단일 큐	독립 큐
채팅방 간 간섭	있음	없음
채팅방별 최대 처리량	전체 합산 20건/분	채팅방당 20건/분

 

 

 채팅방 수가 적고 알림 빈도가 낮다면 단일 큐로도 충분하다. 하지만 채팅방 간 간섭 없이 각 채팅방이 독립적으로 최대 처리량을 활용하고 싶다면, 채팅방별 독립 큐 + 채팅방당 전용 스레드로 가는 것이 적합하다는 생각이 들었다.

 

 

 

 

 

---

최종 구현

 채팅방(chatId) 단위로 큐와 전용 발송 스레드를 관리한다. 같은 채팅방으로 향하는 메시지끼리만 직렬화되고, 서로 다른 채팅방은 독립적으로 동작한다.

 

// 채팅방별 큐
ConcurrentHashMap<String, BlockingQueue<Runnable>> roomQueues = new ConcurrentHashMap<>();
// 채팅방별 스레드
ConcurrentHashMap<String, ExecutorService> roomExecutors = new ConcurrentHashMap<>();

 

 

메시지 전송  로직

send() 호출
  → computeIfAbsent: 첫 호출이면 큐 + 가상 스레드 생성
  → queue.offer(task): 큐에 적재 후 즉시 리턴 (워커 스레드 블록 없음)
  → 가상 스레드: poll(500ms) → task.run() → sleep(3000) → ...

 

 

Graceful Shutdown 추가

애플리케이션이 갑작스럽게 종료되는 상황에서도 큐에 쌓인 메시지가 유실되지 않도록 Graceful Shutdown을 추가했다.

 

boolean running 플래그로 루프를 자연 종료시킨다.

@PreDestroy shutdown()
  → running = false
  → poll(500ms) 타임아웃 → while 루프 탈출
  → drainQueue(): 남은 메시지 순차 전송
  → awaitTermination(): 드레인 완료 대기

 

 

 

최종 구현 코드

@Slf4j
@Service
public class TelegramNotificationService {

    private volatile boolean running = true;
    private final ConcurrentHashMap<String, BlockingQueue<Runnable>> roomQueues = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, ExecutorService> roomExecutors = new ConcurrentHashMap<>();

    public void send(String chatId, String token, String message) {
        BlockingQueue<Runnable> queue = roomQueues.computeIfAbsent(chatId, id -> {
            LinkedBlockingQueue<Runnable> q = new LinkedBlockingQueue<>(500);
            ExecutorService executor = Executors.newSingleThreadExecutor(
                    Thread.ofVirtual().name("telegram-" + id).factory()
            );
            executor.submit(() -> runSender(id, q));
            roomExecutors.put(id, executor);
            return q;
        });
        queue.offer(() -> TelegramApi.send(chatId, token, message));
    }

    private void runSender(String chatId, BlockingQueue<Runnable> queue) {
        while (running) {
            try {
                Runnable task = queue.poll(500, TimeUnit.MILLISECONDS);
                if (task == null) continue;
                task.run();
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("텔레그램 메시지 전송 실패 [chatId={}]", chatId, e);
            }
        }
        drainQueue(chatId, queue);
    }

    private void drainQueue(String chatId, BlockingQueue<Runnable> queue) {
        log.info("텔레그램 잔여 메시지 전송 [chatId={}, count={}]", chatId, queue.size());
        Runnable task;
        while ((task = queue.poll()) != null) {
            try {
                task.run();
            } catch (Exception e) {
                log.error("종료 중 메시지 전송 실패 [chatId={}]", chatId, e);
            }
        }
    }

    @PreDestroy
    public void shutdown() {
        log.info("텔레그램 발송 스레드 종료 시작 - 잔여 메시지 전송 중...");
        running = false;
        roomExecutors.values().forEach(ExecutorService::shutdown);
        roomExecutors.forEach((id, executor) -> {
            try {
                if (!executor.awaitTermination(10, TimeUnit.MINUTES)) {
                    log.warn("텔레그램 발송 타임아웃 [chatId={}]", id);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        log.info("텔레그램 발송 스레드 종료 완료");
    }
}

 

 

 

종료 테스트 로그