전체 글 검색 결과

해당 글 129건

Kafka 공통 메시지 포맷을 위한 제네릭 적용하기

Kafka에서 메시지를 발행(Pub)하고 구독(Sub)할 때, 각 서비스마다 공통된 메시지 구조가 없으면 메시지 생성과 변환을 위한 코드가 중복되어 관리가 어려워지고 복잡성이 증가합니다. 이를 해결하기 위해 제네릭(Generic) 타입을 사용하여 공통 메시지 포맷을 만들었습니다.   제네릭(Generics)제네릭은 Java 5부터 도입된 기능으로, 다양한 타입의 객체들을 다루는 메서드나 클래스에서 컴파일 시에 타입을 미리 지정하지 않고, 유연하게 처리할 수 있게 합니다.  제네릭 장점타입 안정성 증가: 컴파일 시에 타입 오류를 검출하여 런타임 예외를 방지합니다.타입 캐스팅 제거: 불필요한 타입 캐스팅을 줄여 코드를 간결하게 만듭니다.코드 재사용성: 다양한 타입에 대해 동작하는 로직을 하나의 객체를 통해 ..
MQ 2024. 9. 12. 16:06

Kafka Pub CircuitBreaker 적용하기

Pub, Sub, Stream, Kafka와 같이 분산 시스템을 구축하면서 각 서비스에서 다른 서비스를 호출할 때 실패하거나 지연이 되어 해당 영향도가 다른 서비스에 확산되지 않도록 하기 위해 서킷 브레이커를 학습하고 적용하였습니다. 이벤트 요청이 누락되면 안되기 때문에 Kafka Pub 실패 시 아래 AS-IS 플로우처럼 Pub 애플리케이션에서 지속적으로 재시도 호출을 하고 있습니다. Kafka 장애가 복구되기 전까지 같은 오류가 지속적으로 발생하여 Pub 애플리케이션에도 문제가 생길 수 있고 각 외부 서비스에 장애가 전파되기 때문에 TO-BE와 같이 서킷 브레이커를 적용해 봤습니다. AS-IS사용자 → Pub 애플리케이션 → Kafka 장애 → 호출 실패 → 오류 → retry …. → Kafka 복..
MQ 2024. 9. 7. 14:31

Kafka Pub 동적 Batch 관리

현재 사내에서 상품 정보를 변경 시에 처리하는 로직은 세 가지 방식이 있습니다.1. 변경이 일어나 상품정보에 수정이 필요한 데이터들을 수정 처리.2. 스케줄러를 통해 변경된 데이터를 주기적으로 감지 후 처리.3. 전체 색인, 데이터 전체를 동기화 시켜줌. 위 처리방식의 문제점을 해결하기 위해 카프카를 도입하고 3가지 방식을 없애거나 줄이는 방법을 생각했습니다.하지만 3번 방식으로 처리했던 코드를 분석시에 2번과 같은 방법이 여전히 필요했습니다. 예를 들어 아래 전시 상품 프로시저는 3번 방식으로 처리하는 로직이였습니다. (전체 색인, 데이터 전체를 동기화 시켜줌)분석을 해보면 이벤트로 처리 하지 못하는 부분이 생겼습니다.1. 전시 상품 프로시저 분석판매 종료 대상 업체 → 이벤트전시 여부, 판매 상태 →..
MQ 2024. 8. 18. 14:55

Kafka Pub 호출 방안

현재 사내에서 상품 정보를 변경 시에 처리하는 로직은 세 가지 방식이 있습니다.1. 변경이 일어나 상품정보에 수정이 필요한 데이터들을 수정 처리.2. 스케줄러를 통해 변경된 데이터를 주기적으로 감지 후 처리.3. 전체 색인, 데이터 전체를 동기화 시켜줌. 위 처리방식의 문제점을 해결하기 위해 카프카를 도입하고 아래와 같이 Pub, Sub, Stream 3개의 프로젝트를 구성하려 합니다. 현재 사내에서는 이벤트가 발생하는곳은 업체에 제공해주는 내부/외부 관리자 페이지, 배치 프로세스에서 발생하고 있습니다. 위 2가지에서 발생하는 이벤트를 Kafka에 Pub 해주는 방법은 2가지가 있었습니다.1. 관리자 페이지, 배치 프로세스에서 직접 Kafka에 Pub 하기2, 별도 애플리케이션을 통한 Kafka에 Pub..
MQ 2024. 8. 18. 13:31

Kafka 학습하기_컨슈머

[인프런 - 아파치 카프카 애플리케이션 프로그래밍] 컨슈머: 카프카 클러스터에 저장된 레코드를 받아와서 처리하는 애플리케이션 Fetcher: 리더 파티션으로부터 레코드들을 미리 가져와서 대기. poll(): Fetcher에 있는 레코드들을 리턴하는 레코드 ConsumerRecords: 처리하고자 하는 레코드들의 모음. 오프셋이 포함되어있음. 컨슈머 그룹 동일한 역할을 하는 컨슈머들의 묶음. 각 컨슈머 그룹으로부터 격리된 환경에서 안전하게 운영할 수 있도록 도와주는 카프카의 독특한 방식이다. 컨슈머 그룹으로 묶인 컨슈머가 토픽을 구독해서 데이터를 가져갈때, 1개의 파티션은 최대 1개의 컨슈머에 할당이 가능하다. 그리고 1개 컨슈머는 여러개의 파티션에 할당될 수 있다. 이러한 특징으로 컨슈머 그룹의 컨슈머 ..
MQ 2024. 6. 5. 08:00

Kafka 학습하기_프로듀서

[인프런 - 아파치 카프카 애플리케이션 프로그래밍] 프로듀서 애플리케이션ProducerRecord: 프로듀서에서 생성하는 레코드. 오프셋은 미포함. Send: 레코드를 전송 요청 메서드. Partitioner: 어느 파티션으로 전송할지 지정하는 파티셔너. 기본값으로 DefaultPartitioner로 설정됨. Accumulator: 배치로 묶어 전송할 데이터를 모으는 버퍼  프로듀서의 기본 파티셔너 프로듀서API를 사용하면 2개의 파티셔너를 제공한다. UniformStickyPartitioner, RoundRobinPartitioner  2.5.0 버전에서 파티셔너를 지정하지 않는 경우 UniformStickyPartitioner로  설정된다. 메시지 키가 있을 경우 동작 UniformStickyPart..
MQ 2024. 6. 4. 08:00

Kafka 학습하기_토픽, 파티션, 레코드

[인프런 - 아파치 카프카 애플리케이션 프로그래밍] 토픽카프카에서 데이터를 구분하기 위해 사용되는 단위이다. 토픽은 1개 이상의 파티션을 소유하고 있다. 파티션에는 프로듀서가 보낸 데이터들이 들어가 저장되는데 이 데이터들이 들어가 저장되는데 이 데이터를 레코드라고 부른다. 파티션은 자료구조에서 접하는 큐와 비슷하다.(FIFO) 카프카에서는 레코드를 가져가도 삭제하지 않는다. 여러 컨슈머 그룹들이 토픽의 데이터를 여러번 가져갈수있다. 토픽 생성시 파티션이 배치되는 방법round-robin 방식으로 리더 파티션들이 생성된다. 리더 파티션이 있는 브로커와 통신하여 데이터를 주고 받으므로 여러 브로커에 골고루 네트워크 통신을 하게 된다. 데이터가 특정 서버와 통신이 집중되는 현상을 막고 선형 확장을 하여 데이터..
MQ 2024. 6. 2. 01:04

Kafka 학습하기_브로커와 클러스터

[인프런 - 아파치 카프카 애플리케이션 프로그래밍]주키퍼 앙상블- 주키퍼는 분산 애플리케이션을 위한 중앙 집중식 서비스로, 카프카의 메타데이터 관리 및 브로커의 상태를 유지시킴.- 카프카 3.0부터는 주키퍼가 없어도 클러스터 동작 가능.카프카 클러스터- 여러 브로커로 구성된 클러스터로, 주키퍼와 연동하여 메시지를 관리, - 주키퍼는 브로커들의 상태를 감시하고, 장애가 발생하면 클러스터의 균형을 유지하도록 도와줌. 브로커의 역할  - 하나의 서버나 인스턴스 위에서 동작함 - 1개로도 기본 기능 실행 가능 - 3개 이상의 브로커 서버를 1개의 클러스터로 묶어 운영 - 데이터를 안전하게 분산 저장하고, 복제해 장애로부터 가용성 확보 가능 - 컨트롤러 클러스터의 다수 브로커 중 한 대가 컨트롤러의 역할을 한다...
MQ 2024. 6. 1. 17:52

Kafka 학습하기_장단점

[인프런 - 아파치 카프카 애플리케이션 프로그래밍]  작년 11월에 이직하여 현재 근무하고 있는 이커머스 회사에서 서비스를 파악하면서 개선해보고 싶은 내용이 있어 Kafka 학습을 시작하였습니다. 개선점현재 사내에서 상품 정보를 변경 시에 처리하는 로직은 두 가지 방식이 있습니다. 1. 변경이 일어나 상품정보에 수정이 필요한 데이터들을 수정 처리. 2. 스케줄러를 통해 변경된 데이터를 주기적으로 감지 후 처리. 위 두가지 방식의 문제점 1. 단일 프로세스 부하 증가. - 하나의 프로세스에서 여러 작업을 처리하다 보니 시스템의 부하 증가. 이는 성능 저하, 트래픽이 많은 시기에는 다른 서비스에도 영향을 줄 수 있음.2. 비효율 적인 자원 사용. - 스케줄러를 통해 변경된 데이터를 주기적으로 감지하고 처리하..
MQ 2024. 6. 1. 16:16

Elasticsearch CircuitBreaker GC 관련

최근 사내에서 검색엔진인 엘라스틱서치 클러스터 중 하나의 노드가 다운되고 샤드가 옮겨지는 과정에서 검색 API 지연이 발생하였습니다. 노드는 바로 올라와 크게 지연은 있지 않았지만 비정상 종료이기 때문에 해당 서버와 노드의 지표를 찾아보았고 Heap Memory가 가득 차면서 CircuitBreaker가 발생하고 종료된 걸 확인할 수 있었습니다. 노드가 다운되는 시점과 GC Old 영역의 쌓여있던 객체를 정리하는 시점과 동일했기 때문에 해당 과정이 원인이라 생각하게 되었고 GC가 발생하게 된 대용량 색인이나 검색 요청이 있었는지 확인해 보았지만 크게 이상은 없어 정확한 원인을 찾지 못했습니다... 2023.10.26 - [ELK] - Elasticsearch Essential 트러블 슈팅 사례 Elast..
ELK 2023. 10. 26. 16:29