Nellie's Blog

Kafka를 사용할 경우, Producer에서 데이터를 전송할 때 압축률(compression.type)을 지정할 수가 있는데, 자주 쓰는 4가지 압축 방법을 비교하여 포스팅을 해보려고 한다. (매번 헷갈린다..) 자주 쓰는 압축 방법은 gzip, snappy, lz4, zstd 4가지 정도가 있는데, 이 4가지를 테스트해보았다. 파티션이 1개인 Kafka Topic에 건당 10KB 데이터를 전송하여, 압축률 /속도 /CPU 사용량을 비교해보았다.  ※ 테스트 카프카 구성 스펙- 버전 : 3.8.1- 브로커 : 3개- 파티션: 1개 - 카프카 프로듀서 : 자바 API (postman으로 호출)- 사용한 모니터링 도구 : Kafka UI   1. gzip1) 100건 전송 (1MB)Kafka UI에서 좌측..

카프카에서 웹소켓으로 메시지를 전송하여 실시간 처리를 해야한다. 지금 나는 백엔드 개발자인데, 프론트에 내가 만든 mqtt 정보 (url, 인증정보, 포트 등) 을 제공해주어야 한다.  백엔드 단에서는 테스트가 모두 완료되었지만, 제공하기 전에, 내가 직접 vue 프론트 프로젝트를 만들어서 내가 보낸 카프카 메시지를 잘 받아오는지 확인하고 싶었다.  백엔드 간단한 코드와 vue 코드를 직접 작성하고, 테스트 하는 과정을 정리했다.    사용한 기술 및 버전 스프링 부트 : 2.5.4자바 : 1.8카프카  : 3.7.0MQTT : 1.2.5 (vue 에서는 5.9.1)vue  : 3.2.13 백엔드 코드1. 차량 ID 별로 MQTT 토픽 생성하여 전송먼저 카프카 컨슈머에서 데이터를 param으로 받고,ob..

차량의 실시간 위치 데이터 (위도, 경도) 를 MQTT를 사용한 웹소켓으로 대시보드에 실시간으로 변하는 데이터를 보여주었다. 카프카 컨슈머에서 받은 데이터를 바로 MQTT로 메시지 전송하였다. MQTT 프로토콜을 지원하는 메시지 브로커인 EMQX 오픈 소스 브로커를 사용하였다. 실시간으로 데이터를 보여주는 것이 목적이었고, 그 다음으로는 초당 몇건을 보여줄 수 있는지를 테스트 했다.   스프링 부트 버전 : 2.5.4자바 버전 : 1.8카프카 버전 : 3.7.0MQTT 버전 : 1.2.5 pom.xml 의존성을 아래와 같이 받아준다.  org.eclipse.paho org.eclipse.paho.client.mqttv3 1.2.5 org.springframework.integrati..

카프카 브로커를 스케일 아웃 하는 것은 처리량을 늘리기 위한 방법은 아니다. (처리량을 늘리려면 가장 먼저 파티션을 늘려야 한다)카프카는 리더 브로커만 실질적으로 프로듀서와 통신하고 나머지 팔로워 브로커는 복제만 담당하기 때문이다.  하지만 혹시 모를 이슈 상황을 대비해 카프카 브로커를 스케일 아웃 해보았다.현재 카프카 브로커는 3개이며 5개로 스케일 아웃을 하고,  다시 3개로 스케일 인을 해보았다.  카프카 버전 : 3.7.0총 소요시간 : 8시간 (인증세팅이 없으면 1시간도 안걸리는 작업이지만, 카프카 브로커 sasl 인증을 추가해 줄 때 Credential을 자꾸 추가해 주는 실수를 해서... 오래 걸렸다. 한번 주키퍼에 저장된 Credentail은 다시 지정해줄 필요가 없다. 인증 프로세스 숙지..

개발 중인 회사 서비스에서 카프카에서 정제한 데이터를 임시데이터베이스에 저장하고, api로 호출해서 사용자 화면에 리턴해야 하는 기능을 개발했다. 처음엔 카프카 스트림즈에서 제공하는 KTable , RocksDB 를 사용했었다.카프카에서 제공하는 라이브러리라서 의존성만 추가해주면 사용할 수 있어서 편리하긴 하지만,카프카에 종속성이 있기에 (카프카 서버가 죽으면 외부 저장소를 사용할 수가 없다..) 리스크가 있어 외부 데이터베이스인 redis로 바꾸기로 했다. redis 를 적용하는 과정을 정리해보았다. springboot 버전 : 2.5.4redis 버전 : 3.3.1총 적용 소요시간 : 1시간 1. redis 서버 설치1) 도커로 설치docker pull redis2) 방화벽 설정firewall-cmd..

카프카 스트림즈로 필터링된 토픽을 컨슈밍하는 컨슈머를 만들고, 그 컨슈머에서 받은 데이터를 DB로 insert하도록 하는 기능을 구현했다.  카프카 커넥트로 하려고 했으나, 번거로워서 직접 컨슈머 코드 안에서 dao를 사용하여 db에 접근하여 insert 하도록 했다.  전체 코드이다.  프로듀서import org.apache.kafka.clients.producer.*;import java.util.Properties;public class KafkaProducerTest { /** * SASL 인증을 위한 JAAS Template */ private static final String JAAS_TEMPLATE = "org.apache.kafka.common.security.s..

kafka1, kafka2, kafka3 중에 하나씩 브로커를 죽여보고 메시지를 전송해보려고 한다. 업무에 kafka 를 적용하여 운영할 예정이라, 브로커 서버가 죽었을 때 메시지가 정상적으로 전송 되는지 고 가용성 테스트를 해보기로 했다.  목차1. directTest 토픽의 partition 3들의 복제 상태 확인2. 프로듀서 메시지 전송 확인3. 브로커 하나씩 죽여보기4. 브로커 하나씩 다시 살리기 결론 ✅ 1. directTest 토픽의 partition 3들의 복제 상태 확인현재 토픽의 복제 상태를 확인하는 명령어kafka-topics.sh --bootstrap-server kafka1:9092 --topic directTest --describe --command-config /opt/bitn..

목차 원래는 토픽 생성 시에 설정 값으로 준다replication-factor를 설정하는 것은 토픽.sh에서 토픽을 생성할 때 설정 값으로 주는 것이다. 아래처럼….$ kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 2 --partitions 10 --topic MyTopic RF를 변경해보자그런데 나는 replication factor를 3으로 변경하고 싶었다. 브로커가 3이기 때문에!일단 토픽을 describe 명령어를 써서 상세 조회 해보자. 1. RF 상세 조회로 현재 RF 값 확인kafka-topics.sh --bootstrap-server kafka1:9092 --topic directTest --d..

로컬 VM에 Docker를 설치하고, 그 안에 카프카를 설치해서 콘솔 테스트까지 해보는 과정을 정리했다.  목차1. Docker 설치  1) yum 패키지 설치 2) 도커 설치 3) 도커 실행 4) docker-compose 설치 2. Docker Hub에서 카프카 및 카프카UI 이미지 내려받기 1) 카프카 이미지 내려받기 2) 주키퍼 이미지 내려받기 3) 카프카 UI 이미지 내려받기 3. docker-compose.yml 설정 1) docker-compose.yml 작성 2) VM에서 포트포워딩 해주기 4. Kafka Cluster 실행 1) docker-compose 실행 2) docker-compose 실행 확인 5. Kafka Console Test 1) Container 내부 쉘 접속 2) To..