Algorithm In Architecture #2 (Apache Kafka)
1. Message Queue
Message Queue 는 기존 레거시부터 최근까지 계속 사용되고 있는 방식입니다.
지원하는 프로토콜(MQTT, AMQP, JMS)에 따라서 분류되기도 하며 메모리 기반/디스크 기반으로 나누어지기도 합니다.
또한 Message를 처리하는 방식에 따라서 Push / Pull 방식으로 나누기도 합니다.
그리고 Queue 의 정상적인 동작을 돕기 위해서 Broker가 개입하는 것이 일반적입니다.
그러나 다양한 제품 가운데 공통적인 것이 있으니 결국은 Queue 라는 것입니다.
FIFO(First In First Out)의 메시지 처리가 필요하다는 전제로 각 시스템의 특징과 요구사항에 따라 맞춰서 솔루션을 선택하면 됩니다.
2. Message Queue 를 사용하는 이유는?
특히 Queue 는 비동기식 메시지 처리에서 많이 사용되는데 그 이유는 다음과 같습니다.
- 어플리케이션을 구현할 때 메시지가 어디서 오는지(Source), 어디로 보내야 하는지(Target) 신경쓸 필요가 없다.
- Producer는 메시지를 작성하는 데에만 집중하고 메시지 전달에 관련된 (목적지, 순서, QoS 등) 항목들에 대해서는 MQ에 모두 위임한다.
- Consumer 역시 해당 메시지를 어떻게 소모하여 처리하는 지에 대해서만 집중하면 된다.
- Producer 와 Consumer가 보다 각자의 역할에 집중할 수 있게 도와준다.
3. 자료구조 Queue
특정 자료구조에 대한 정리나 코드연습은 Algorithm 카테고리에서 자세히 소개할 예정이기 때문에 간단히 살펴봅니다.
이해를 돕기 위해서 array를 이용해서 작성한 Queue Sample입니다.
class Queue {
private static int front, rear, capacity;
private static int queue[];
Queue(int c)
{
front = rear = 0;
capacity = c;
queue = new int[capacity];
}
static void queueEnqueue(int data)
{
// check queue is full or not
if (capacity == rear) {
return;
}
// insert element at the rear
else {
queue[rear] = data;
rear++;
}
return;
}
static void queueDequeue()
{
// if queue is empty
if (front == rear) {
return;
}
else {
for (int i = 0; i < rear - 1; i++) {
queue[i] = queue[i + 1];
}
if (rear < capacity)
queue[rear] = 0;
rear--;
}
return;
}
}
-queueEnqueue
rear 는 현재 데이터를 넣어야하는 위치의 index 입니다.
만약에 capacity 만큼 꽉 차있다면 더 이상 데이터를 넣을 수 없습니다.
데이터를 넣는 것이 가능하다면 삽입 후 index를 변경합니다.
-queueDequeue
front는 데이터가 삭제되는 위치의 index입니다.
front == rear 라면 비어있는 상태입니다.
데이터를 제거하는 것이 가능하다면 제거 후 index를 변경합니다.
4. Queue 사용시 주의점
위의 sample소스를 통해서 Queue의 특징을 이해하셨다면 주의점도 파악하실 수 있습니다.
- Queue는 무한한 용량이 아니다. 적절히 소모가 이루어지지 않는다면 Queue의 모든 용량을 소모하게 되고 유실이 발생한다.
- Queue에는 많은 쓰레드들이 동시에 접근할 수 있다. enqueue/dequeue 동작은 thread-safe 해야 합니다.
Java에서는 이를 위해서 concurrent 패키지를 제공합니다.
- 서로 다른 Queue에서는 순서보장이 되지 않습니다. (당연한 말씀)
Computer Science의 기초인 DS & Algorithm은 반드시 알고 있는 것이 좋습니다.
5. Apache Kafka
많은 Message Queue 중에서 Apache Kafka는 대용량 실시간 처리에 특화된 아키텍처 설계를 통하여 우수한 TPS를 보장합니다.
기본적으로 pub-sub 구조로 동작하며, scale out과 high availability를 확보하는 것으로 유명합니다.
아키텍처상 다양한 특징이 있지만 이번 글에서는 자료구조 & 알고리즘 측면에서 살펴보겠습니다.
5-A. Multi Partition
Kafka가 분산처리가 가능한 것은 Multi Partition이 가능하기 때문입니다.
대용량의 데이터가 들어올 수록 병렬처리를 통해서 속도를 확보할 수 있는 구조를 가지고 있습니다.
그렇다면 단점은 없을까요?
서로 다른 Partition에 대해서는 순서를 보장하지 않습니다.(다른 큐로 보시면 됩니다.)
예를 들어서 Partition 0 내의 메시지들끼리, Partition 1 내의 메시지들끼리 순서를 보장하지만 Partition 0과 1사이에는 순서를 보장할 수 없습니다.
반드시 순서를 보장해야 하는 케이스가 존재한다면?
- Partition 을 하나만 사용하던지
- Partitioner를 Custom 하게 작성하여 순서가 보장되어야 하는 메시지들은 같은 Partition으로 보내도록 합니다.
Topic과 혼동하시면 안됩니다. 특정 Topic내에 여러 Partition이 존재하며 Broker를 통해서 각 Topic별 Parition 의 저장정보와 복제 등이 이루어집니다.
5-B. Partitioner
Kafka의 Default Partitioner는 modulo 연산을 사용하여 파티션을 나누게 되어 있습니다.
Hash 알고리즘을 잘 작성하여 데이터를 고르게 분산하면 병렬처리 성능을 더 끌어올릴 수 있습니다.
간단한 Hash Function의 예제입니다.
private int hashFunction(K key){
int val=0;
for(int i=0; i<key.toString().length();i++){
val+= (key.toString().charAt(i)) % size;
}
return val % size;
}일반적으로 우리가 사용하는 HashMap에서도 같은 Hash값이 자주 발생할 경우 특정 row가 길어지게 되어 chaining hash table을 구현하게 되는데, 이 때 worst case로 특정 bucket에 데이터가 집중되어 메모리를 효율적으로 사용하지 못하는 경우가 발생합니다.
Kafka에서도 partition을 적절하게 분산하지 못할 경우 이러한 문제가 발생할 수 있습니다.
6. 정리
- Apach Kafka는 자료구조 Queue와 동일합니다.
- 병렬처리를 위해서 multi partiton을 지원합니다. 여러 개의 Queue 를 동시에 사용한다고 생각하면 됩니다.
- Partitioner를 어떻게 구현하는가에 따라서 partiton이 결정됩니다.
- FIFO보장이 필요한 단위에 대해서는 같은 partiton 이 적용되도록 구현합니다.
0. 참조
https://sarc.io/index.php/miscellaneous/1615-message-queue-mq
https://ko.wikipedia.org/wiki/%ED%81%90_(%EC%9E%90%EB%A3%8C_%EA%B5%AC%EC%A1%B0)
https://monsieursongsong.tistory.com/
https://en.wikipedia.org/wiki/Hash_table
https://www.geeksforgeeks.org/array-implementation-of-queue-simple/