Middleware/Kafka&RabbitMQ 2019. 3. 30. 00:45

Kafka - Spring Cloud Stream Kafka Streams API(스프링 클라우드 스트림 카프카 스트림즈 API)

kafka is a distributed streaming platform

이번 포스팅은 Spring Cloud Stream 환경에서의 kafka Streams API입니다. 물론 이전 포스팅들에서 자바코드로 카프카 스트림즈를 다루어봤지만 이번에는 스프링 클라우드 스트림즈 환경에서 진행합니다. 카프카 스트림즈에 대한 설명은 이전 포스팅에서 진행하였기에 개념적인 설명은 하지 않고 코드레벨로 다루어보겠습니다. 혹시나 카프카 스트림즈를 모르시는 분이 있으시다면 아래 링크를 참조하시길 바랍니다.

지금부터 진행될 예제 및 설명은 모두 Spring Cloud Stream Reference를 참조하였습니다. 번역 중 오역이 있을 수 있습니다.

▶︎▶︎▶︎2019/03/26 - [Kafka&RabbitMQ] - Kafka - Kafka Streams API(카프카 스트림즈)

 

Kafka - Kafka Streams API(카프카 스트림즈)

Kafka - Kafka Streams API(카프카 스트림즈) 카프카는 대규모 메시지를 저장하고 빠르게 처리하기 위해 만들어진 제품이다. 초기 사용 목적과는 다른 뛰어난 성능에 일련의 연속된 메시지인 스트림을 처리하는..

coding-start.tistory.com

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
<?xml version="1.0" encoding="UTF-8"?>
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.3.RELEASE</version>
        <relativePath /> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>kafka-streams-exam</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka-streams-exam</name>
    <description>Demo project for Spring Boot</description>
 
    <properties>
        <java.version>1.8</java.version>
        <spring-cloud.version>Greenwich.SR1</spring-cloud.version>
    </properties>
 
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream</artifactId>
        </dependency>
 
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-stream-test-support</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
 
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
 
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
 
</project>
 
http://colorscripter.com/info#e" target="_blank" style="color:#e5e5e5; text-decoration:none">Colored by Color Scripter
 http://colorscripter.com/info#e" target="_blank" style="text-decoration:none; color:white">cs

의존성을 추가해줍니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@SpringBootApplication
@EnableBinding(KStreamProcessor.class)
public class WordCountProcessorApplication {
 
    @StreamListener("input")
    @SendTo("output")
    public KStream<?, WordCount> process(KStream<?, String> input) {
        return input
                .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
                .groupBy((key, value) -> value)
                .windowedBy(TimeWindows.of(5000))
                .count(Materialized.as("WordCounts-multi"))
                .toStream()
                .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end()))));
    }
 
    public static void main(String[] args) {
        SpringApplication.run(WordCountProcessorApplication.class, args);
    }
}
 cs

해당 코드를 레퍼런스에서 소개하는 카프카 스트림즈의 예제코드이다 토픽에서 메시지를 받아서 적절하게 데이터를 가공하여 특정 토픽으로 데이터를 내보내고 있다 내부 로직은 마치 Java1.8의 Stream와 비슷한 코드같이 보인다 메소드를 보면 매개변수와 반환타입이 모두 KStream객체다 나가고 들어노는 객체의 타입은 하나로 고정되어있으므로 비지니스로직에 조금이라도 더 집중할 수 있을 것 같다.

 

Configuration Option

Kafka Streams Properties는 모두 spring.cloud.stream.kafka.streams.binder 접두어가 붙는다. 

 

  • brokers - broker URL
  • zkNodes - zookeeper URL
  • serdeError - 역 직렬화 오류 처리 유형. logAndContinue,logAndFail,sendToDlq 
  • applicationId - 애플리케이션에서 사용하는 카프카 스트림 아이디값. 모든 카프카 스트림의 아이디값은 유일해야한다.

Producer Properties

  • keySerde - key serde, 키의 직렬화 옵션(default value = none)
  • valueSerde -  value serde, 값의 직렬화 옵션 (default value = none)
  • userNativeEncoding - native 인코딩 활성화 플래그(default value = false)

Consumer Properties

  • keySerde - key serde, 키의 역직렬화 옵션(default value = none)
  • valueSerde - value serde, 값의 역직렬화 옵션(default value = none)
  • materializedAs - KTable 타입을 사용할 때 상태저장소를 구체화한다.(default value = none)
  • useNativeDecoding - native 디코드 활성화 플래그(default value = false)
  • dlqName - DLQ 토픽 이름(default value = none)

기타 다른 설정들은 레퍼런스 참고하시길 바랍니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
/**
 * in,out channel defined
 * 즉, 카프카 토픽에 메시지를 쓰는 발신 채널과
 * 카프카 토픽에서 메시지를 읽어오는 수신 채널을 정의해주는 것이다.
 * 꼭 @Input & @Output 어노테이션을 하나씩 넣을 필요는 없다.
 * 필요한 채널수만큼 정의가능하다.
 * 
 * 런타임동안에는 스프링이 구현체를 제공해준다.
 * @author yun-yeoseong
 *
 */
public interface ExamProcessor {
    String INPUT = "exam-input";
    String OUTPUT = "exam-output";
    String OUTPUT2 = "exam-output2";
    
    /**
     * @Input 어노테이션 설명
     * 메시지 시스템에서 메시지를 읽어오는 입력채널 
     * 매겨변수로 채널이름을 넣을 수 있다. 넣지 않으면 기본으로
     * 메소드 이름으로 들어간다.
     * @return
     */
    @Input(INPUT)
    SubscribableChannel inboundChannel();
    
    @Input("streams-input")
    KStream<?, String> inboundChannel2();
    
    @Input("streams-input2")
    KStream<?, String> inboundChannel3();
    /**
     * @Output 어노테이션 설명
     * 메시지 시스템으로 메시지를 보내는 출력채널
     * 매겨변수로 채널이름을 넣을 수 있다. 넣지 않으면 기본으로
     * 메소드 이름으로 들어간다.
     * @return
     */
    @Output(OUTPUT)
    MessageChannel outboundChannel();
    
    @Output(OUTPUT2)
    MessageChannel outboundChannel2();
    
    @Output("streams-output")
    KStream<?, String> outboundChannel3();
    
}
 
 cs

카프카 스트림즈를 스프링 클라우드 스트림에서 사용하려면 input&output을 KStream을 반환타입으로 채널을 정의한다. 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@Slf4j
@EnableBinding(ExamProcessor.class)
@SpringBootApplication
public class KafkaStreamsExamApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(KafkaStreamsExamApplication.class, args);
    }
    
    @StreamListener(target= ExamProcessor.INPUT,condition="headers['producerName']=='yeoseong'")
    public void receive(@Payload Exam exam) {
        log.info("Only Header value yeoseong = {}",exam.toString());
    }
    
    @StreamListener("streams-input")
    @SendTo("streams-output")
    public KStream<?, String> streams(KStream<?, String> input){
        return input.flatMapValues(new ValueMapper() {
            @Override
            public Object apply(Object value) {
                // TODO Auto-generated method stub
                String valueStr = (String)value;
                System.out.println(valueStr);
                return Arrays.asList(valueStr.split(" "));
            }
        });
    }
    
    @StreamListener("streams-input2")
    public void streams2(KStream<?, String> input) {
        System.out.println("streams2");
        input.foreach( (key,value) -> System.out.println(value));
    }
}
 
 cs

위에서 보이듯이 sink processor라면 그 밑으로 더 이상 processor가 붙지 않으므로 void 반환타입으로 데이터를 입력받아 적절한 처리를 하면된다. 하지만 밑으로 스트림이 더 붙는 스트림들은 반환타입으로 KStream을 반환해야한다. 지금까지 간단하게 Kafka Streams API를 Spring Cloud Stream에서 사용해보았다. kafka streams DSL의 자세한 문법은 곧 포스팅할 카프카 스트림즈 API에 대한 글을 참고하시거나 카프카 레퍼런스를 참고하시면 될듯합니다. 

 

저작자표시

'Middleware > Kafka&RabbitMQ' 카테고리의 다른 글

Apache Kafka - Kafka(카프카)란 ? 분산 메시징 플랫폼 - 1  (0) 2019.07.06
Kafka - Kafka Stream API(카프카 스트림즈) - 2  (0) 2019.03.30
Kafka - Spring cloud stream kafka(스프링 클라우드 스트림 카프카)  (0) 2019.03.28
Kafka - Kafka Streams API(카프카 스트림즈) - 1  (2) 2019.03.26
Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI  (2) 2019.03.24
posted by 여성게
:
Middleware/Kafka&RabbitMQ 2019. 3. 28. 16:41

Kafka - Spring cloud stream kafka(스프링 클라우드 스트림 카프카)

Kafka - Spring cloud stream kafka(스프링 클라우드 스트림 카프카)


이전 포스팅까지는 카프카의 아키텍쳐, 클러스터 구성방법, 자바로 이용하는 프로듀서,컨슈머 등의 글을 작성하였다. 이번 포스팅은 이전까지 작성된 지식을 바탕으로 메시징 시스템을 추상화한 구현체인 Spring Cloud Stream을 이용하여 카프카를 사용하는 글을 작성하려고 한다. 혹시라도 카프카에 대해 아직 잘모르는 사람들이 이 글을 본다면 이전 포스팅을 한번 참고하고 와도 좋을 것같다.(이번에 작성하는 포스팅은 Spring Cloud stream 2.0 레퍼런스 기준으로 작성하였다.)

그리고 이번 포스팅에서 진행하는 모든 예제는 카프카를 미들웨어로 사용하는 예제이고, 카프카는 클러스터를 구성하였다.


▶︎▶︎▶︎Spring Cloud Stream v2.0 Reference

▶︎▶︎▶︎2019/03/12 - [Kafka&RabbitMQ] - Kafka - Kafka(카프카)의 동작 방식과 원리

▶︎▶︎▶︎2019/03/13 - [Kafka&RabbitMQ] - Kafka - Kafka(카프카) cluster(클러스터) 구성 및 간단한 CLI사용법

▶︎▶︎▶︎2019/03/16 - [Kafka&RabbitMQ] - Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI

▶︎▶︎▶︎2019/03/24 - [Kafka&RabbitMQ] - Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI

▶︎▶︎▶︎2019/03/26 - [Kafka&RabbitMQ] - Kafka - Kafka Streams API(카프카 스트림즈)


스프링 클라우드 스트림을 이용하면 RabbitMQ,Kafka와 같은 미들웨어 메시지 시스템과는 종속적이지 않게 추상화된 방식으로 메시지 시스템을 이용할 수 있다.(support RabbitMQ,Kafka) 아래 그림은 스프링 클라우드 스트림의 애플리케이션 모델이다.

스프링 클라우드 스트림 애플리케이션과 메시지 미들웨어 시스템은 직접 붙지는 않는다. 중간에 스프링 클라우드 스트림이 제공하는 바인더 구현체를 중간에 두고 통신을 하기 때문에 애플리케이션에서는 미들웨어 독립적으로 추상화된 방식으로 개발 진행이 가능하다. 그리고 애플리케이션과 바인더는 위의 그림과 같이 inputs, outputs 채널과 통신을 하게 된다.

  • Binder : 외부 메시징 시스템과의 통합을 담당하는 구성 요소입니다.
  • Binding(input/output) : 외부 메시징 시스템과 응용 프로그램 간의 브리지 (대상 바인더에서 생성 한 메시지 생성자 및 소비자 ).
  • Middleware : RabbitMQ, Kafka와 같은 메시지 시스템.

바인더 같은 경우는 스프링이 설정에서 읽어 미들웨어에 해당하는 바인더를 구현체로 제공해준다. 물론 RabbitMQ, Kafka를 동시에 사용 가능하다. 그리고 바인딩 같은 경우는 기본으로 Processor(input,output),Source(output),Sink(input)라는 바인딩을 인터페이스로 제공한다. 이러한 스프링 클라우드 스트림을 이용하여 작성한 완벽히 동작하는 코드의 예제이다.


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@SpringBootApplication
@EnableBinding(Processor.class)
public class MyApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(MyApplication.class, args);
    }
 
    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public String handle(String value) {
        System.out.println("Received: " + value);
        return value.toUpperCase();
    }
}
cs


이 코드는 Processor라는 바인딩 채널을 사용하고, handle이라는 메소드에서 Processor의 input 채널에서 메시지를 받아서 값을 한번 출력하고 그대로 output 채널로 메시지를 보내는 코드이다.


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public interface Sink {
 
  String INPUT = "input";
 
  @Input(Sink.INPUT)
  SubscribableChannel input();
 
}
public interface Source {
 
  String OUTPUT = "output";
 
  @Output(Source.OUTPUT)
  MessageChannel output();
 
}
 
public interface Processor extends Source, Sink {}
cs


스프링 클라우드 스트림에서 기본적으로 제공하는 바인딩 채널이다. 만약 이 채널들 이외에 채널을 정의하고 싶다면 위와 같이 인터페이스로 만들어주고, @EnableBinding에 매개변수로 넣어주면된다.(매개변수는 여러개의 인터페이스를 받을 수 있다.)


1
2
3
4
5
6
7
8
9
10
11
public interface Barista {
 
    @Input
    SubscribableChannel orders();
 
    @Output
    MessageChannel hotDrinks();
 
    @Output
    MessageChannel coldDrinks();
}
cs

또한 채널 바인딩 어노테이션(@Input,@Output) 매개변수로 채널이름을 넣어줄 수 있다.

1
@EnableBinding(value = { Orders.class, Payment.class })
cs


스프링 클라우드 스트림에서 바인딩 가능한 채널타입은 MessageChannel(inbound)와 SubscribableChannel(outbound) 두개이다.

1
2
3
4
5
6
public interface PolledBarista {
 
    @Input
    PollableMessageSource orders();
    . . .
}
cs


지금까지는 이벤트 기반 메시지이지만 위처럼 Pollable한 채널을 바인딩 할 수 있다.


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@Autowire
private Source source
 
public void sayHello(String name) {
    source.output().send(MessageBuilder.withPayload(name).build());
}
 
@Autowire
private MessageChannel output;
 
public void sayHello(String name) {
    output.send(MessageBuilder.withPayload(name).build());
}
 
@Autowire
@Qualifier("myChannel")
private MessageChannel output;
cs


정의한 채널에 대한 @Autowired하여 직접 사용도 가능하다. 그리고 @Qualifier 어노테이션을 이용해 다중 채널이 정의되어 있을 경우 특정한 채널을 주입받을 수 있다.


@StreamListener 사용

스프링 클라우드 스트림은 다른 org.springframework.messaging의 어노테이션도 같이 사용가능하다.(@Payload,@Headers,@Header)


1
2
3
4
5
6
7
8
9
10
11
@EnableBinding(Sink.class)
public class VoteHandler {
 
  @Autowired
  VotingService votingService;
 
  @StreamListener(Sink.INPUT)
  public void handle(Vote vote) {
    votingService.record(vote);
  }
}
cs


1
2
3
4
5
6
7
8
9
10
11
12
13
14
@Component
@Slf4j
public class KafkaListener {
    
    @StreamListener(ExamProcessor.INPUT)
    @SendTo(ExamProcessor.OUTPUT2)
    public Exam listenMessage(@Payload Exam payload,@Header("contentType") String header) {
        log.info("input message = {} = {}",payload.toString(),header);
        return payload;
    }
}
 
=>결과 : input message = Exam(id=id_2, describe=test message !) = application/json
 
cs


또한 @SendTo 어노테이션을 추가로 붙여서 메시지를 수신하고 어떠한 처리를 한 후에 메시지를 출력채널로 내보낼 수 있다.


1
2
3
4
5
6
7
8
9
10
11
12
@EnableBinding(Processor.class)
public class TransformProcessor {
 
  @Autowired
  VotingService votingService;
 
  @StreamListener(Processor.INPUT)
  @SendTo(Processor.OUTPUT)
  public VoteResult handle(Vote vote) {
    return votingService.record(vote);
  }
}
cs


바로 위의 코드는 즉, 인바운드 채널에서 메시지를 받아서 그대로 해당 데이터를 리턴하여 아웃바운드 채널에 내보내는 예제이다. 실제로는 메소드 내에 비지니스 로직이 들어가 적절히 데이터 조작이 있을 수 있다.


@StreamListener for Content-based routing

조건별로 @StreamListener 주석처리가 된 인입채널 메소드로 메시지를 유입시킬 수 있다. 하지만 이 기능에는 밑에와 같은 조건을 충족해야한다.


  • 반환값이 있으면 안된다.
  • 개별 메시지 처리 메소드여야한다.

조건은 어노테이션 내의 condition 인수에 SpEL 표현식에 의해 지정된다. 그리고 조건과 일치하는 모든 핸들러는 동일한 스레드에서 호출되며 핸들러에 대한 호출이 발생하는 순서는 가정할 수 없다.


1
2
3
4
5
6
7
8
9
10
11
12
13
14
@EnableBinding(Sink.class)
@EnableAutoConfiguration
public static class TestPojoWithAnnotatedArguments {
 
    @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bogey'")
    public void receiveBogey(@Payload BogeyPojo bogeyPojo) {
       // handle the message
    }
 
    @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bacall'")
    public void receiveBacall(@Payload BacallPojo bacallPojo) {
       // handle the message
    }
}
cs


위에서 얘기한 것과 같이 조건별로 메시지를 분기하는 경우 해당 채널로 @StreamListener된 메소드들은 모두 반환값이 void여야한다.(Sink.INPUT) 만약 위에서 위는 void, 아래에는 반환값이 있는 메소드로 선언한다면 예외가 발생한다.



Error Handler

Spring Cloud Stream은 오류 처리를 유연하게 처리하는 메커니즘을 제공한다. 오류 처리에는 크게 두가지가 있다.

  • application : 사용자 정의 오류처리이며, 애플리케이션 내에서 오류 처리를 진행한다.
  • system : 오류 처리가 기본 메시징 미들웨어의 기능에 따라 달라진다.


Application Error Handler


애플리케이션 레벨 오류 처리에는 두 가지 유형이 있다. 오류는 각 바인딩 subscription에서 처리되거나 전역 오류 핸들러가 모든 바인딩 subscription의 오류를 처리한다. 각 input 바인딩에 대해, Spring Cloud Stream은 <destinationName>.<groupName>.errors 설정으로 전용 오류 채널을 생성한다.


1
spring.cloud.stream.bindings.input.group=myGroup
cs


컨슈머 그룹을 설정한다.(만약 그룹명을 지정하지 않았을 경우 익명으로 그룹이 생성된다.)


1
2
3
4
5
6
7
8
9
@StreamListener(Sink.INPUT) // destination name 'input.myGroup'
public void handle(Person value) {
    throw new RuntimeException("BOOM!");
}
//만약 Sink.INPUT의 input 채널이름이 input이고, 해당 채널의 consumer group이 myGroup이라면
//해당 채널 단독의 에러 처리기의 inputChannel 매개변수의 값은 아래와 같다.("input.myGroup.errors")
@ServiceActivator(inputChannel = "input.myGroup.errors") //channel name 'input.myGroup.errors'
public void error(Message<?> message) {
    System.out.println("Handling ERROR: " + message);
}
cs


위와 같은 코드를 작성하면 Sink.INPUT.myGroup 채널의 전용 에러채널이 생기는 것이다. 하지만 이렇게 채널별이 아닌 전역 에러 처리기를 작성하려면 아래와 같은 코드를 작성하면 된다.


1
2
3
4
@StreamListener("errorChannel")
public void error2(Message<?> message) {
    log.error("Global Error Handling !");
}
cs


System Error Handling


System 레벨의 오류 처리는 오류가 메시징 시스템에 다시 전달되는 것을 의미하며, 모든 메시징 시스템이 동일하지는 않다. 즉, 바인더마다 기능이 다를 수 있다. 내부 오류 처리기가 구성되어 있지 않으면 오류가 바인더에 전파되고 바인더가 해당 오류를 메시징 시스템에 전파한다. 메시징 시스템의 기능에 따라 시스템은 메시지를 삭제하고 메시지를 다시 처리하거나 실패한 메시지를 DLQ로 보낼 수 있다. 해당 내용은 뒤에서 더 자세히 다룬다.


바인딩 정보 시각화


1
2
3
4
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
cs


의존성을 추가해준다.


1
management.endpoints.web.exposure.include=bindings
cs


application.propertis에 해당 설정을 추가해준 후에, host:port/actuator/bindings 를 호출하면 바인딩된 정보를 JSON형태로 받아볼 수 있다.



Binder Configuration Properties


Spring Auto configuration을 사용하지 않고 사용자 정의 바인더를 등록할때 다음 등록 정보들을 사용할 수 있다. 이 속성들은 모두 org.springframework.cloud.stream.config.BinderProperties 패키지에 정의되어있다. 그리고 해당 설정들은 모두 spring.cloud.stream.binders. 접두어가 붙는다. 밑에서 설명할 설정들은 모두 접두어가 생략된 형태이므로 실제로 작성할때는 접두어를 붙여준다.


  • type

바인더의 타입을 지정한다.(rabbit,kafka)

  • inheritEnvironment

애플리케이션 자체 환경을 상속하는지 여부

기본값 : true

  • environment

바인더 환경을 사용자가 직접 정의하는데 사용할 수 있는 설정이다.

기본값 : empty

  • defaultCandidate

바인더 구성이 기본 바인더로 간주되는지 또는 명시적으로 참조할 때만 사용할 수 있는지 여부 이 설정을 사용하면 기본 처리를 방해하지 않고 바인더 구성이 가능하다.

기본값 : true



Common Binding Properties


binding 설정입니다. 해당 설정은 spring.cloud.stream.bindings.<channelName> 접두어가 붙는다. 밑에서 설명할 설정은 모두 접두어가 생략된 설정이므로, 직접 애플리케이션을 설정할때에는 꼭 접두어를 붙여줘야한다.


  • destination

해당 채널을 메시지 시스템 토픽과 연결해주는 설정이다. 채널이 consumer로 바인딩되어 있다면, 여러 대상에 바인딩 될 수 있으며 대상 이름은 쉼표로 구분된 문자열이다.

  • group

컨슈머 그룹명설정이다. 해당 설정은 인바운드 바인딩에만 적용되는 설정이다.

기본값 : null

  • contentType

메시지의 컨텐츠 타입이다.

기본값 : null

  • binder

사용될 바인더를 설정한다.

기본값 : null



Consumer Properties

  • concurrency

인바운드 소비자의 동시성

기본값 : 1.

  • partitioned

컨슈머가 파티션된 프로듀서로부터 데이터를 수신하는지 여부입니다.

기본값 : false.

  • headerMode

none으로 설정하면 입력시 헤더 구문 분석을 사용하지 않습니다. 기본적으로 메시지 헤더를 지원하지 않으며 헤더 포함이 필요한 메시징 미들웨어에만 유효합니다. 이 옵션은 원시 헤더가 지원되지 않을 때 비 Spring Cloud Stream 애플리케이션에서 데이터를 사용할 때 유용합니다. 로 설정 headers하면 미들웨어의 기본 헤더 메커니즘을 사용합니다. 로 설정 embeddedHeaders하면 메시지 페이로드에 헤더가 포함됩니다.

기본값 : 바인더 구현에 따라 다릅니다.

  • maxAttempts

처리가 실패하면 다시 메시지를 처리하는 시도 횟수 (첫 번째 포함). 1로 설정하면 다시 메시지 처리를 시도하지 않는다.

기본값 : 3.

  • backOffInitialInterval

다시 시도 할 때 백 오프 초기 간격입니다.

기본값 : 1000.

  • backOffMaxInterval

최대 백 오프 간격.

기본값 : 10000.

  • backOffMultiplier

백 오프 승수입니다.

기본값 : 2.0.

  • instanceIndex

0보다 큰 값으로 설정하면이 소비자의 인스턴스 색인을 사용자 정의 할 수 있습니다 (다른 경우spring.cloud.stream.instanceIndex). 음수 값으로 설정하면 기본값은로 설정됩니다.(카프카의 경우 autoRebalence 설정이 false 일 경우)

기본값 : -1.

  • instanceCount

0보다 큰 값으로 설정하면이 소비자의 인스턴스 수를 사용자 정의 할 수 있습니다 (다른 경우 spring.cloud.stream.instanceCount). 음수 값으로 설정하면 기본값은로 설정됩니다.(카프카의 경우 autoRebalence 설정이 false 일 경우)

기본값 : -1.


Producer Properties

  • partitionKeyExpression

아웃 바운드 데이터를 분할하는 방법을 결정하는 SpEL 식입니다. set 또는 ifpartitionKeyExtractorClass가 설정되면이 채널의 아웃 바운드 데이터가 분할됩니다.partitionCount효과가 있으려면 1보다 큰 값으로 설정해야합니다. 

기본값 : null.

  • partitionKeyExtractorClass

PartitionKeyExtractorStrategy구현입니다. set 또는 if partitionKeyExpression가 설정되면이 채널의 아웃 바운드 데이터가 분할됩니다. partitionCount효과가 있으려면 1보다 큰 값으로 설정해야합니다. 

기본값 : null.

  • partitionSelectorClass

PartitionSelectorStrategy구현입니다.

기본값 : null.

  • partitionSelectorExpression

파티션 선택을 사용자 정의하기위한 SpEL 표현식. 

기본값 : null.

  • partitionCount

파티셔닝이 사용 가능한 경우 데이터의 대상 파티션 수입니다. 제작자가 분할 된 경우 1보다 큰 값으로 설정해야합니다. 카프카에서는 힌트로 해석됩니다. 이것보다 크고 대상 항목의 파티션 수가 대신 사용됩니다.

기본값 : 1.


Content-Type


Spring Cloud Stream은 contentType에 대해 세 가지 메커니즘을 제공한다.


  • Header 

contentType 자체를 헤더로 제공한다.

  • binding

spring.cloud.stream.bindings.<inputChannel>.content-type 설정으로 타입을 설정한다.

  • default

contentType이 명시적으로 설정되지 않은 경우 기본 application/json 타입으로 적용한다.


위의 순서대로 우선순위 적용이 된다.(Header>bindings>default) 만약 메소드 반환 타입이 Message 타입이면 해당 타입으로 메시지를 수신하고, 만약 일반 POJO로 반환타입이 정의되면 컨슈머에서 해당 타입으로 메시지를 수신한다.





Apache Kafka Binder


1
2
3
4
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
        </dependency>
cs


의존성을 추가해준다.


Kafka Binder Properties

  • spring.cloud.stream.kafka.binder.brokers

Kafka 바인더가 연결될 브로커 목록이다.

기본값 : localhost

  • spring.cloud.stream.kafka.binder.defaultBrokerPort

brokers 설정에 포트 정보가 존재하지 않는다면 해당 호스트 리스트들이 사용할 포트를 지정해준다.

기본값 : 9092

  • spring.cloud.stream.kafka.binder.configuration

바인더로 작성된 모든 클라이언트에 전달 될 클라이언트 속성(프로듀서,컨슈머)의 키/값 맵이다. 이러한 속성은 프로듀서와 컨슈머 모두가 사용한다.

기본값 : empty map

  • spring.cloud.stream.kafka.binder.headers

바인더에 의해 전송되는 사용자 지정 헤더 목록이다.

기본값 : null

  • spring.cloud.stream.kafka.binder.healthTimeout

파티션 정보를 얻는 데 걸리는 시간(초). 

기본값 : 10

  • spring.cloud.stream.kafka.binder.requiredAcks

브로커에서 필요한 ack수이다.(해당 설명은 이전 포스팅에 설명되어있다.)

기본값 : 1

  • spring.cloud.stream.kafka.binder.minPartitionCount

autoCreateTopics,autoAddPartitions true를 설정했을 경우에만 적용되는 설정이다. 바인더가 데이터를 생성하거나 소비하는 주제에 대해 바인더가 구성하는 최소 파티션 수이다.

기본값 : 1

  • spring.cloud.stream.kafka.binder.replicationFactor

autoCreateTopics true를 설정했을 경우에 자동 생성된 토픽 복제요소 수이다.

기본값 : 1

  • spring.cloud.stream.kafka.binder.autoCreateTopics

true로 설정하면 토픽이 존재하지 않을 경우 자동으로 토픽을 만들어준다. 만약 false로 설정되어있다면 미리 토픽이 생성되어 있어야한다.

기본값 : true

  • spring.cloud.stream.kafka.binder.autoAddPartitions

true로 설정하면 바인더가 필요할 경우 파티션을 추가한다. 예를 들어 사용자의 메시지 유입이 증가하여 컨슈머를 추가하여 파티션수가 하나더 늘었다고 가정하자. 그러면 기존의 토픽의 파티션 수는 증설한 파티션의 총수보다 작을 것이고, 이 설정이 true라면 바인더가 자동으로 파티션수를 증가시켜준다.

기본값 : false



Kafka Consumer Properties


spring.cloud.stream.kafka.bindings.<inputChannel>.consumer 접두어가 붙는다.

  • autoRebalanceEnabled

파티션 밸런싱을 자동으로 처리해준다.

기본값 : true

  • autoCommitOffset

메시지가 처리되었을 경우, 오프셋을 자동으로 커밋할지를 설정한다.

기본값 : true

  • startOffset

새 그룹의 시작 오프셋이다. earliest, latest

기본값 : null(==eariest)

  • resetOffsets
consumer의 오프셋을 startOffset에서 제공한 값으로 재설정할지 여부
기본값 : false


Kafka Producer Properties

  • bufferSize

kafka 프로듀서가 전송하기 전에 일괄 처리하려는 데이터 크기이다.

기본값 : 16384

  • batchTimeout

프로듀서가 메시지를 보내기 전에 동일한 배치에 더 많은 메시지가 누적될 수 있도록 대기하는 시간. 예를 들어 버퍼사이즈가 꽉 차지 않았을 경우 얼마나 기다렸다고 메시지 처리할 것인가를 정하는 시간이다.

기본값 : 0


Partitioning with the Kafka Binder


순서가 중요한 메시지가 있을 경우가 있다. 이럴 경우에는 어떠한 키값을 같이 포함시켜 메시지를 발신함으로써 하나의 파티션에만 데이터를 보내 컨슈머쪽에서 데이터의 순서를 정확히 유지하여 데이터를 받아올 수 있다.


1
2
3
#partition-key-expression
spring.cloud.stream.bindings.exam-output.producer.partition-key-expression=headers['partitionKey']
spring.cloud.stream.bindings.exam-output.producer.partition-count=4
cs


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
    @Autowired
    private ExamProcessor processor;
    
    public void sendMessage(Exam exam,String partitionId) {
        log.info("Sending Message = {}",exam.toString());
        
        MessageChannel outputChannel = processor.outboundChannel();
        
        outputChannel.send(MessageBuilder
                                    .withPayload(exam)
                                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                                    .setHeader("partitionKey", partitionId)
                                    .build());
    }
 
    @StreamListener(ExamProcessor.INPUT)
    public void listenMessage(@Payload Exam payload,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int header) {
        log.info("input message = {} partition_id= {}",payload.toString(),header);
    }
cs


application.properties에 파티션키로 사용할 헤더의 키값을 설정하고, 프로듀서 쪽에서 키값을 포함시켜 데이터를 보낸 후에 컨슈머쪽에서 어떠한 파티션에서 데이터를 받았는지 확인해본다. 키값을 동일하게 유지한채 메시지를 보내면 하나의 파티션에서만 메시지를 받아온다. 그러나 파티션 키값을 적용하지 않고 메시지를 보내면 컨슈머가 받아오는 파티션 아이디는 계속해서 변경이 된다.


이번 포스팅은 내용이 조금 길어져서 여기까지만 작성하고 다음 포스팅에서 Spring Cloud Stream 기반에서 사용하는 kafka stream API에 대해 포스팅할 것이다. 이번 포스팅에서는 많은 설명들이 레퍼런스에 비해 빠져있다. 아는 내용은 작성하고 모르는 내용은 포함시킬경우 틀릴 가능성이 있기 때문에 레퍼런스와 비교해 내용에 차이가 있다. 혹시나 더 많은 설명이 필요하다면 직접 레퍼런스를 보면 될것 같다.

저작자표시

'Middleware > Kafka&RabbitMQ' 카테고리의 다른 글

Kafka - Kafka Stream API(카프카 스트림즈) - 2  (0) 2019.03.30
Kafka - Spring Cloud Stream Kafka Streams API(스프링 클라우드 스트림 카프카 스트림즈 API)  (0) 2019.03.30
Kafka - Kafka Streams API(카프카 스트림즈) - 1  (2) 2019.03.26
Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI  (2) 2019.03.24
Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI  (0) 2019.03.16
posted by 여성게
:
Middleware/Kafka&RabbitMQ 2019. 3. 26. 00:07

Kafka - Kafka Streams API(카프카 스트림즈) - 1

Kafka - Kafka Streams API(카프카 스트림즈)


카프카는 대규모 메시지를 저장하고 빠르게 처리하기 위해 만들어진 제품이다. 초기 사용 목적과는 다른 뛰어난 성능에 일련의 연속된 메시지인 스트림을 처리하는 데도 사용이 되기 시작했다. 이러한 스트림을 카프카는 Kafka Streams API를 통해 제공한다.


설명하기 앞서 우선 스트림 프로세싱과 배치 프로세싱의 차이점이란 무엇일까? 스트림 프로세싱(Stream Processing)은 데이터들이 지속적으로 유입되고 나가는 과정에서 이 데이터에 대한 일련의 처리 혹은 분석을 수행하는 것을 의미한다. 즉, 스트림 프로세싱은 실시간 분석(Real Time Analysis)이라고 불리기도 한다. 스트림 프로세싱과는 대비되는 개념으로 배치(Batch)처리 또는 정적 데이터(Data-at-rest)처리를 들 수 있다. 배치 및 정적 데이터 처리란 위의 스트림 프로세싱과는 다르게 데이터를 한번에 특정 시간에 처리한다라는 특징이 있다. 주로 사용자의 요청이 몰리지 않는 새벽 시간대에 많이 수행하기도 한다.(물론 무조건 그렇다고는 할 수 없다) 그렇지만 사실 뭐가 좋고 나쁨은 이야기 할 수 없다. 스트림과 배치의 서로의 장단점이 있기 때문이다. 하지만 요즘은 역시나 실시간 데이터 처리가 각광받고 있는 사실은 숨길 수 없다.


<특정 이벤트 발생시 바로바로 애플리케이션에서 데이터처리를 하고 있다.>


그렇다면 스트림 프로세싱의 장점은 무엇이 있을까? 우선은 애플리케이션이 이벤트에 즉각적으로 반응을 하기 때문에 이벤트 발생과 분석,조치에 있어 거의 지연시간이 발생하지 않는다. 또한 항상 최신의 데이터를 반영한다라는 특징이 있다. 그리고 데이터를 어디에 담아두고 처리하지 않기 때문에 대규모 공유 데이터베이스에 대한 요구를 줄일 수 있고, 인프라에 독립적인 수행이 가능하다.



상태 기반과 무상태 스트림 처리

스트림 프로세싱에는 상태 기반과 무상태 스트림 처리가 있다. 차이점이란 실시간 데이터 처리를 위하여 이전에 분석된 데이터의 결과가 필요한지이다. 이렇게 상태를 유지할 스트림 프로세싱은 이벤트를 처리하고 그 결과를 저장할 상태 저장소가 필요하다. 이와는 반대로 무상태 스트림 처리는 이전 스트림의 처리 결과와 관계없이 현재 데이터로만 처리를 한다.



카프카 스트림즈(Kafka Streams API)



카프카 스트림즈는 카프카에 저장된 데이터를 처리하고 분석하기 위해 개발된 클라이언트 라이브러리이다. 카프카 스트림즈는 이벤트 시간과 처리 시간을 분리해서 다루고 다양한 시간 간격 옵션을 지원하기에 실시간 분석을 간단하지만 효율적으로 진행할 수 있다. 우선 카프카 스트림즈에 대해 설명하기 전에 용어 정리를 할 필요가 있을 것같다.


용어 

설명 

스트림(Stream) 

스트림은 카프카 스트림즈 API를 사용해 생성된 토폴로지로, 끊임없이 전달되는 데이터 세트를 의미한다. 스트림에 기록되는 단위는 key-value 형태이다. 

스트림 처리 애플리케이션(Stream Processing Application) 

카프카 스트림 클라이언트를 사용하는 애플리케이션으로서, 하나 이상의 프로세서 토폴로지에서 처리되는 로직을 의미한다. 프로세서 토폴로지는 스트림 프로세서가 서로 연결된 그래프를 의미한다. 

스트림 프로세서(Stream Processor) 

프로세서 토폴로지를 이루는 하나의 노드를 말하여 여기서 노드들은 프로세서 형상에 의해 연결된 하나의 입력 스트림으로부터 데이터를 받아서 처리한 다음 다시 연결된 프로세서에 보내는 역할을 한다. 

 소스 프로세서(Source Processor)

위쪽으로 연결된 프로세서가 없는 프로세서를 말한다. 이 프로세서는 하나 이상의 카프카 토픽에서 데이터 레코드를 읽어서 아래쪽 프로세서에 전달한다. 

싱크 프로세서(Sink Processor) 

토폴로지 아래쪽에 프로세서 연결이 없는 프로세서를 뜻한다. 상위 프로세서로부터 받은 데이터 레코드를 카프카 특정 토픽에 저장한다. 



카프카 스트림즈 아키텍쳐

카프카 스트림즈에 들어오는 데이터는 카프카 토픽의 메시지이다. 각 스트림 파티션은 카프카의 토픽 파티션에 저장된 정렬된 메시지이고, 해당 메시지는 key-value형태이다. 또한 해당 메시지의 키를 통해 다음 스트림(카프카 토픽)으로 전달된다. 위의 그림에서 보듯 카프카 스트림즈는 입력 스트림의 파티션 개수만큼 태스크를 생성한다. 각 태스크에는 입력 스트림(카프카 토픽) 파티션들이 할당되고, 이것은 한번 정해지면 입력 토픽의 파티션에 변화가 생기지 않는 한 변하지 않는다. 마지 컨슈머 그룹 안의 컨슈머들이 각각 토픽 파티션을 점유하는 것과 비슷한 개념이다.



자바를 이용한 간단한 파이프 예제 프로그램

지금부터 진행할 예제는 이전 포스팅에서 구성했던 카프카 클러스터 환경에서 진행한다. 만약 클러스터 구성이 되어 있지않다면 밑의 링크를 참조해 구성하면 될것 같다.

스프링 부트 애플리케이션 하나를 생성한다. 그리고 필요한 라이브러리 의존성을 추가한다. 로그를 사용하기 위하여 롬복 라이브러리를 추가하였다. 혹시 이클립스 환경에서 롬복을 사용하는 방법을 알고 싶다면 밑의 링크를 참조하자.

▶︎▶︎▶︎2019/02/02 - [Java&Servlet] - Mac OS - Eclipse & Lombok(롬복 사용방법)


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.4</version>
            <scope>provided</scope>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>2.0.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.12</artifactId>
            <version>2.0.1</version>
        </dependency>
cs


이번에 만들 프로그램은 간단히 한쪽 토픽에 입력된 값을 다른 쪽 토픽으로 옮기는 역할을 수행한다. 


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
 
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
 
import lombok.extern.slf4j.Slf4j;
 
@Slf4j
public class Pipe {
    
    public static void main(String[] args) {
        
        /*
         * 카프카 스트림 파이프 프로세스에 필요한 설정값
         * StreamsConfig의 자세한 설정값은
         * https://kafka.apache.org/10/documentation/#streamsconfigs 참고
         */
        Properties props = new Properties();
        //카프카 스트림즈 애플리케이션을 유일할게 구분할 아이디
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-pipe");
        //스트림즈 애플리케이션이 접근할 카프카 브로커정보
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094,localhost:9095");
        //데이터를 어떠한 형식으로 Read/Write할지를 설정(키/값의 데이터 타입을 지정) - 문자열
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        
        //데이터의 흐름으로 구성된 토폴로지를 정의할 빌더
        final StreamsBuilder builder = new StreamsBuilder();
        
        //streams-*input에서 streams-*output으로 데이터 흐름을 정의한다.
        /*
         * KStream<String, String> source = builder.stream("streams-plaintext-input");
           source.to("streams-pipe-output");
         */
        builder.stream("streams-plaintext-input").to("streams-pipe-output");
        
        //최종적인 토폴로지 생성
        final Topology topology = builder.build();
        
        //만들어진 토폴로지 확인
        log.info("Topology info = {}",topology.describe());
        
        final KafkaStreams streams = new KafkaStreams(topology, props);
        
        try {
          streams.start();
          System.out.println("topology started");
      
        } catch (Throwable e) {
          System.exit(1);
        }
        System.exit(0);
    }
}
 
cs


우선 Properties 객체를 이용하여 카프카 스트림즈가 사용할 설정값을 입력한다. 우선 StreamsConfig.APPLICATION_ID_CONFIG는 카프카 클러스터 내에서 스트림즈 애플리케이션을 구분하기 위한 유일한 아이디 값을 설정하기 위한 값이다. 그리고 그 밑에 카프카 클러스터 구성이 된 인스턴스를 리스트로 작성한다. 또한 위에서 설명했듯이 카프카 스트림즈는 키-값 형태로 데이터의 흐름이 이루어지기에 해당 키-값을 어떠한 타입으로 직렬화 할 것인지 선택한다. 현재는 String 타입으로 지정하였다. 그 다음은 스트림 토폴로지를 구성해준다. 토폴로지는 StreamsBuilder 객체를 통하여 구성한다. 위의 소스는 "streams-plaintext-input"이라는 토픽에서 데이터를 읽어 "streams-pipe-output"이라는 토픽으로 데이터를 보내는 토폴로지를 구성하였다. 최종적으로 StreamsBuilder.build()를 호출하여 토폴로지 객체를 만든다. 위의 Topology.describe()를 호출하면 해당 토폴로지가 어떠한 구성으로 이루어졌는지 로그로 상세하게 볼 수 있다.


22:07:24.234 [main] INFO com.kafka.exam.streams.Pipe - Topology info = Topologies:

   Sub-topology: 0

    Source: KSTREAM-SOURCE-0000000000 (topics: [streams-plaintext-input])

      --> KSTREAM-SINK-0000000001

    Sink: KSTREAM-SINK-0000000001 (topic: streams-pipe-output)

      <-- KSTREAM-SOURCE-0000000000


그 다음 KafkaStreams 객체를 생성한다. 매개변수로 토폴로지와 설정 객체를 전달한다.  마지막으로 KafkaStreams.start();를 호출하여 카프카 스트림을 시작한다. 해당 예제에서는 명시하지 않았지만 스트림즈를 종료하려면 close() 해주는 코드도 삽입해야한다.



카프카가 제공하는 콘솔 프로듀서,컨슈머를 이용하여 위의 스트림즈 코드를 수행해보았다. input 토픽에서 스트림즈 애플리케이션이 메시지를 꺼내서 output 토픽으로 잘 전달한것을 볼 수 있다.(>"hi hello") 컨슈머 실행시 세팅한 설정은 직관적으로 해석 가능하므로 따로 설명하지는 않는다.


방금은 단순히 메시지를 받아 다른 쪽 토픽으로 전달만 하는 예제이지만 다음 해볼 예제는 중간에 데이터를 처리하여 output 토픽으로 보내는 행 분리 예제 프로그램이다.


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
 
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;
 
import lombok.extern.slf4j.Slf4j;
 
@Slf4j
public class LineSplit {
    public static void main(String[] args) {
        /*
         * 카프카 스트림 파이프 프로세스에 필요한 설정값
         * StreamsConfig의 자세한 설정값은
         * https://kafka.apache.org/10/documentation/#streamsconfigs 참고
         */
        Properties props = new Properties();
        //카프카 스트림즈 애플리케이션을 유일할게 구분할 아이디
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-linesplit");
        //스트림즈 애플리케이션이 접근할 카프카 브로커정보
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094,localhost:9095");
        //데이터를 어떠한 형식으로 Read/Write할지를 설정(키/값의 데이터 타입을 지정) - 문자열
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        
        //데이터의 흐름으로 구성된 토폴로지를 정의할 빌더
        final StreamsBuilder builder = new StreamsBuilder();
        
        KStream<String, String> source = builder.stream("streams-plaintext-input");
        source.flatMapValues(value -> Arrays.asList(value.split("\\W+")))
            .to("streams-linesplit-output");
        
        //최종적인 토폴로지 생성
        final Topology topology = builder.build();
        
        //만들어진 토폴로지 확인
        log.info("Topology info = {}",topology.describe());
        
        final KafkaStreams streams = new KafkaStreams(topology, props);
 
        try {
          streams.start();
          System.out.println("topology started");
        } catch (Throwable e) {
          System.exit(1);
        }
        System.exit(0);
    }
}
 
cs


설정 값은 동일하다 하지만 토폴로지에 로직이 추가된 것을 볼 수 있다. 우선 "streams-plaintext-input" 토픽으로 들어온 메시지를 KStream객체로 만들어준다. 그리고 flatMapValues()를 이용하여 데이터를 공백 단위로 쪼개는 것을 볼 수 있다. 여기서 flatMapValues()를 쓴 이유는 여기서 전달되는 value는 하나의 스트링 객체인데 이것을 List<String>으로 변환을 하였다. 이것을 flatMapValues()를 이용해 리스트의 각 요소를 스트링으로 flat하는 메소드이다. 즉, 메시지를 받아서 flatMapValues()를 통해 새로운 데이터 값이 만들어 졌다. 그러면 .to()에 전달되는 데이터 스트림은 공백단위로 짤린 단어 스트링이 전달될 것이다. 이전 예제와는 다르게 단순 데이터 전달만이 아니라 전달 이전에 적절한 처리를 하나 추가 해준 것이다. 만약 키와 값 모두를 새롭게 만들어서 사용하고 싶다면 flatMap() 메소드를 이용하면 된다. 그 밖에 메소드들을 알고 싶다면 공식 홈페이지를 참고하길 바란다.


22:49:08.941 [main] INFO com.kafka.exam.streams.LineSplit - Topology info = Topologies:

   Sub-topology: 0

    Source: KSTREAM-SOURCE-0000000000 (topics: [streams-plaintext-input])

      --> KSTREAM-FLATMAPVALUES-0000000001

    Processor: KSTREAM-FLATMAPVALUES-0000000001 (stores: [])

      --> KSTREAM-SINK-0000000002

      <-- KSTREAM-SOURCE-0000000000

    Sink: KSTREAM-SINK-0000000002 (topic: streams-linesplit-output)

      <-- KSTREAM-FLATMAPVALUES-0000000001


토폴로지 정보를 보아도 중간에 Processor라는 이름으로 하나의 처리 프로세스가 추가된 것을 볼 수 있다.



hi hello 라는 메시지를 보냈는데 hi / hello 로 짤려서 출력된 것을 볼 수 있다. 


필자는 챗봇을 개발하고 있는데 카프카 스트림즈를 이용하여 간단히 사용자 질문 로그들을 분석하여 형태소 분리된 형태로 데이터를 저장할 수도 있을 것 같아서 간단히 예제로 짜보았다. NoriAnalyzer는 필자가 Nori 형태소 분석기를 이용한 간단히 형태소분석 유틸 클래스를 만든것이다.


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
 
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;
 
import lombok.extern.slf4j.Slf4j;
import rnb.analyzer.nori.analyzer.NoriAnalyzer;
 
@Slf4j
public class LineSplit {
    public static void main(String[] args) {
        /*
         * 카프카 스트림 파이프 프로세스에 필요한 설정값
         * StreamsConfig의 자세한 설정값은
         * https://kafka.apache.org/10/documentation/#streamsconfigs 참고
         */
        Properties props = new Properties();
        //카프카 스트림즈 애플리케이션을 유일할게 구분할 아이디
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-linesplit");
        //스트림즈 애플리케이션이 접근할 카프카 브로커정보
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094,localhost:9095");
        //데이터를 어떠한 형식으로 Read/Write할지를 설정(키/값의 데이터 타입을 지정) - 문자열
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        
        //데이터의 흐름으로 구성된 토폴로지를 정의할 빌더
        final StreamsBuilder builder = new StreamsBuilder();
        //Nori 형태소 분석기를 이용한 유틸클래스
        NoriAnalyzer analyzer = new NoriAnalyzer();
        
        KStream<String, String> source = builder.stream("streams-plaintext-input");
        source.flatMapValues(value -> Arrays.asList(analyzer.analyzeForString(value).split(" ")))
            .to("streams-linesplit-output");
        
        //최종적인 토폴로지 생성
        final Topology topology = builder.build();
        
        //만들어진 토폴로지 확인
        log.info("Topology info = {}",topology.describe());
        
        final KafkaStreams streams = new KafkaStreams(topology, props);
 
        try {
          streams.start();
          System.out.println("topology started");
        } catch (Throwable e) {
          System.exit(1);
        }
    }
}
 
cs



이렇게 추후에는 형태소분리된 모든 단어가 아니라 특정 명사만 추출하여 사용자들의 질문중 가장 많이 포함된 명사들을 뽑아내 데이터를 분석할 수도 있을 것같다.


지금까지의 예제는 바로 무상태 스트림 프로세싱이다. 다음 예제는 이전 데이터 처리에 대한 상태를 참조하여 데이터를 처리하는 단어 빈도수 세기 프로그램 예제이다. 


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import java.util.Arrays;
import java.util.Locale;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
 
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.state.KeyValueStore;
 
import lombok.extern.slf4j.Slf4j;
import rnb.analyzer.nori.analyzer.NoriAnalyzer;
 
@Slf4j
public class WordCounter {
    
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094,localhost:9095");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
      
        final StreamsBuilder builder = new StreamsBuilder();
        
        NoriAnalyzer analyzer = new NoriAnalyzer();
      
        KStream<String, String> source = builder.stream("streams-plaintext-input");
        source.flatMapValues(value -> Arrays.asList(analyzer.analyzeForString(value).split(" ")))
                //return하는 value가 키가 된다.
                .groupBy(new KeyValueMapper<String, String, String>() {
                  @Override
                  public String apply(String key, String value) {
                    log.info("key = {},value = {}",key,value);
                    return value;
                  }
                })
                //count()를 호출하여 해당 키값으로 몇개의 요소가 있는지 체크한다. 그리고 해당 데이터를 스토어에(KeyValueStore<Bytes,byte[]> counts-store) 담고

  //변경이 있는 KTable에 대해서만 결과값을 리턴한다.

                .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"))
                .toStream()
                .to("streams-wordcount-output", Produced.with(Serdes.String(), Serdes.Long()));
      
        final Topology topology = builder.build();
        System.out.println(topology.describe());
      
        final KafkaStreams streams = new KafkaStreams(topology, props);
      
        try {
          streams.start();
        } catch (Throwable e) {
          System.exit(1);
        }
    }
}
 
cs


설정값은 동일하다. 조금 달라진 것은 데이터처리부이다. 우선은 입력받은 데이터를 형태소분리하여 토큰 단위로 스트림을 만든다. 그리고 groupBy를 이용하여 토큰값을 키로 하여 스트림을 그룹핑한다. 그 다음 count()를 통해 같은 키값으로 몇개의 요소가 있는 지를 KeyValueStore에 담고, 만약 키값스토어의 값이 변경이 있을 때만 다운스트림으로 내려준다.(key == null은 무시한다) 처음에는 해당 단어:빈도수가 출력이 되지만 이후에는 이 단어가 또 들어오지 않는다면 출력되지 않고 해당 단어가 들어오면 키값 저장소에 빈도수가 변경이 되므로 다운스트림으로 내려준다. 이말은 무엇이냐면 즉, 키값스토어가 이전 데이터 처리의 상태를 담고 있는 저장소가 되어 이후 데이터에서 참조되어 사용되는 것이다.(사실 더 자세한 것은 API문서를 봐야할 듯하다.)




일부 출력이 안된 것들은 토픽의 버퍼가 원하는 용량에 도달하지 못해 아직 출력하지 못한 데이터들이다. 이렇게 상태를 유지하여 스트림 프로세스를 구성할 수도 있다. 


지금까지 아주 간단하게 카프카 스트림즈 API에 대해 다루었다. 사실 실무에서 이용할 수 있을 만큼의 예제는 아니지만 카프카가 이렇게 실시간 데이터 스트림에도 이용될 수 있다는 것을 알았다면 추후에 API문서를 참조하여 카프카를 응용할 수 있을 것같다.

저작자표시

'Middleware > Kafka&RabbitMQ' 카테고리의 다른 글

Kafka - Spring Cloud Stream Kafka Streams API(스프링 클라우드 스트림 카프카 스트림즈 API)  (0) 2019.03.30
Kafka - Spring cloud stream kafka(스프링 클라우드 스트림 카프카)  (0) 2019.03.28
Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI  (2) 2019.03.24
Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI  (0) 2019.03.16
Kafka - Kafka(카프카) cluster(클러스터) 구성 및 간단한 CLI사용법  (0) 2019.03.13
posted by 여성게
:
Middleware/Kafka&RabbitMQ 2019. 3. 24. 13:54

Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI

Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI


이전 포스팅에서 kafka producer를 java 소스기반으로 예제를 짜보았습니다. 이번 포스팅은 kafka consumer를 java 소스로 다루어보려고 합니다.

Kafka Producer(카프카 프로듀서)가 메시지를 생산해서 카프카의 토픽으로 메시지를 보내면 그 토픽의 메시지를 가져와서 소비(consume)하는 역할을 하는 애플리케이션, 서버 등을 지칭하여 컨슈머라고 한다. 컨슈머의 주요 기능은 특정 파티션을 관리하고 있는 파티션 리더에게 메시지를 가져오기 요청을 하는 것이다. 각 요청은 컨슈머가 메시지 오프셋을 명시하고 그 위치로부터 메시지를 수신한다. 그래서 컨슈머는 가져올 메시지의 위치를 조정할 수 있고, 필요하다면 이미 가져온 데이터도 다시 가져올 수 있다. 이렇게 이미 가져온 메시지를 다시 가져올 수 있는 기능은 여타 다른 메시지 시스템(래빗엠큐,RabbitMQ)들은 제공하지 않는 기능이다.(내부 구동방식이 다른 이유도 있음) 최근의 메시지큐 솔루션 사용자들에게 이러한 기능은 필수가 되고 있다. 

카프카에서 컨슈머라고 불리는 컨슈머는 두 가지 종류가 있는데, 올드 컨슈머(Old Consumer)와 뉴 컨슈머(New Consumer)이다. 두 컨슈머의 큰 차이점은 오프셋에 대한 주키퍼 사용 유무이다. 구 버전의 카프카에서는 컨슈머의 오프셋을 주키퍼의 znode에 저장하는 방식을 지원하다가 카프카 0.9버전부터 컨슈머의 오프셋 저장을 주키퍼가 아닌 카프카의 도픽에 저장하는 방식으로 변경되었다. 아마 성능상의 이슈때문에 오프셋 저장 정책을 바꾼것 같다. 두가지 방식을 특정 버전이전까지는 지원하겠지만 아마 추후에는 후자(뉴 컨슈머)의 방식으로 변경되지 않을까싶다. 코드 레벨로 카프카 컨슈머를 다루기전 카프카 컨슈머의 주요 옵션을 본다.



컨슈머 주요 옵션(Consumer option)

-bootstrap.servers : 카프카 클러스터에 처음 연결을 하기 위한 호스트와 포트 정보로 구성된 리스트 정보이다.


-fetch.min.bytes : 한번에 가져올 수 있는 최소 데이터 사이즈이다. 만약 지정한 사이즈보다 작은 경우, 요청에 대해 응답하지 않고 데이터가 누적될 때까지 기다린다.


-group.id : 컨슈머가 속한 컨슈머 그룹을 식별하는 식별자이다.


-enable.auto.commit : 백그라운드에서 주기적으로 오프셋을 자동 커밋한다.


-auto.offset.reset : 카프카에서 초기 오프셋이 없거나 현재 오프셋이 더 이상 존재하지 않은 경우에 다음 옵션으로 리셋한다.

1)earliest : 가장 초기의 오프셋값으로 설정

2)latest : 가장 마지막의 오프셋값으로 설정

3)none : 이전 오프셋값을 찾지 못하면 에러를 발생


-fetch.max.bytes : 한번에 가져올 수 있는 최대 데이터 사이즈


-request.timeout.ms : 요청에 대해 응답을 기다리는 최대 시간


-session.timeout.ms : 컨슈머와 브로커사이의 세션 타임 아웃시간. 브로커가 컨슈머가 살아있는 것으로 판단하는 시간(기본값 10초) 만약 컨슈머가 그룹 코디네이터에게 하트비트를 보내지 않고 session.timeout.ms이 지나면 해당 컨슈머는 종료되거나 장애가 발생한 것으로 판단하고 컨슈머 그룹은 리밸런스(rebalance)를 시도한다. session.timeout.ms는 하트비트 없이 얼마나 오랫동안 컨슈머가 있을 수 있는지를 제어하는 시간이며, 이 속성은 heartbeat.interval.ms와 밀접한 관련이 있다. session.timeout.ms를 기본값보다 낮게 설정하면 실패를 빨리 감지 할 수 있지만, GC나 poll 루프를 완료하는 시간이 길어지게 되면 원하지 않게 리밸런스가 일어나기도 한다. 반대로는 리밸런스가 일어날 가능성은 줄지만 실제 오류를 감지하는 데 시간이 오래 걸릴 수 있다.


-hearbeat.interval.ms : 그룹 코디네이터에게 얼마나 자주 KafkaConsumer poll() 메소드로 하트비트를 보낼 것인지 조정한다. session.timeout.ms와 밀접한 관계가 있으며 session.timeout.ms보다 낮아야한다. 일반적으로 1/3 값정도로 설정한다.(기본값 3초)


-max.poll.records : 단일 호출 poll()에 대한 최대 레코드 수를 조정한다. 이 옵션을 통해 애플리케이션이 폴링 루프에서 데이터를 얼마나 가져올지 양을 조정할 수 있다.


-max.poll.interval.ms : 컨슈머가 살아있는지를 체크하기 위해 하트비트를 주기적으로 보내는데, 컨슈머가 계속해서 하트비트만 보내고 실제로 메시지를 가져가지 않는 경우가 있을 수도 있다. 이러한 경우 컨슈머가 무한정 해당 파티션을 점유할 수 없도록 주기적으로 poll을 호출하지 않으면 장애라고 판단하고 컨슈머 그룹에서 제외한 후 다른 컨슈머가 해당 파티션에서 메시지를 가져갈 수 있게한다.


-auto.commit.interval.ms : 주기적으로 오프셋을 커밋하는 시간


-fetch.max.wait.ms : fetch.min.bytes에 의해 설정된 데이터보다 적은 경우 요청에 응답을 기다리는 최대 시간


기타 많은 설정들이 있다. 혹시나 인증과 ssl 등의 다양한 설정들을 알고 싶다면 공식 홈페이지를 참고하길 바란다.(https://kafka.apache.org/documentation/#consumerconfigs)



콘솔 컨슈머로 메시지 가져오기


카프카는 기본적으로 콘솔로 메시지를 가져올 수 있는 명령어를 제공한다. 우선 진행하기 앞서 예제로 진행하려는 환경은 이전 포스팅에서 구성해보았던 카프카 클러스터 환경에서 진행한다. 만약 카프카 클러스터 환경을 구성해본적이 없다면 밑의 링크를 참고하기 바란다.


▶︎▶︎▶︎2019/03/13 - [Kafka&RabbitMQ] - Kafka - Kafka(카프카) cluster(클러스터) 구성 및 간단한 CLI사용법


클러스터 환경 구축 후에 예제로 토픽하나를 생성했다는 가정하게 진행한다.



Kafka console producer로 접속하여 메시지를 보내보면 위의 컨슈머로 메시지가 전달 될 것이다. 이러한 컨슈머를 실행할 때는 항상 컨슈머 그룹이라는 것이 필요하다. 토픽의 메시지를 가져오기 위한 kafka-console-consumer.sh 명령어를 실행하면서 추가 옵션으로 컨슈머 그룹 이름을 지정해야 하는데, 만약 추가 옵션을 주지 않고 실행한 경우에는 자동으로 console-consumer-xxxxx(숫자)로 컨슈머 그룹이 생성된다.



필자가 예제로 생성한 그룹아이디도 보인다. 이렇게 그룹아이디 리스트를 볼 수 있는 명령어도 제공한다. 혹시나 모르시는 분들을 위해 토픽 리스트를 볼 수 있는 명령어도 있다.



보면 필자가 사용하고 있는 카프카 버전은 뉴 컨슈머를 이용하는 것을 알 수 있다. 바로 메시지 오프셋을 저장하기 위한 __consumer_offsets가 토픽에 존재하기 때문이다.



또한 컨슈머를 실행시킬 때에 위에서 설명한 것과 같이 그룹아이디 값 옵션을 추가하여 실행시킬 수도 있다.



자바코드를 이용한 컨슈머


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import java.util.Arrays;
import java.util.Properties;
import java.time.Duration;
 
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
 
import lombok.extern.slf4j.Slf4j;
 
/**
 * Kafka Consumer Exam
 * @author yun-yeoseong
 *
 */
@Slf4j
public class KafkaBookConsumer1 {
    
    public static Properties init() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094,localhost:9095");
        props.put("group.id", "exam-consumer-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.offset.reset", "latest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }
    
    /**
     * 컨슈머예제
     */
    public static void consume() {
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(init());){
            //토픽리스트를 매개변수로 준다. 구독신청을 한다.
            consumer.subscribe(Arrays.asList("test-topic"));
            while (true) {
              //컨슈머는 토픽에 계속 폴링하고 있어야한다. 아니면 브로커는 컨슈머가 죽었다고 판단하고, 해당 파티션을 다른 컨슈머에게 넘긴다.
              ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
              for (ConsumerRecord<String, String> record : records)
                log.info("Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
            }
        } finally {}
    }
        
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        consume();
    }
 
}
 
cs


소스에 대해 간단히 설명하면 우선 컨슈머 옵션들을 Properties 객체로 정의한다. 오프셋 리셋 옵션을 lastest로 주어 토픽의 가장 마지막부터 메시지를 가져온다. 메시지의 키와 값의 역직렬화 옵션을 문자열로 해준다. 이렇게 컨슈머 옵션을 설정하고 해당 옵션값을 이용하여 KafkaConsumer 객체를 생성해준다. 해당 객체는 사용이 후에 꼭 close 해줘야 하기 때문에 try문 안에 객체 생성코드를 넣어놔 자동으로 자원반납을 할 수 있게 해주었다. 그리고 해당 컨슈머 객체로 토픽에 대해 구독신청을 해준다. 구독 대상이 되는 토픽은 리스트로 여러개 지정가능하다. 이것 또한 카프카의 장점중 하나이다.(하나의 컨슈머가 여러 메시지큐를 구독하는 것)

그리고 해당 컨슈머는 계속해서 폴링한다. 무한 루프로 계속해서 폴링하지 않으면 컨슈머가 종료된 것으로 간주되어 컨슈머에 할당된 파티션은 다른 컨슈머에게 전달되고 다른 컨슈머에 의해 메시지가 컨슘된다. poll() 메소드의 매개변수는 타임아웃 주기이다. 이 타임아웃주기는 데이터가 컨슈머 버퍼에 없다면 poll()은 얼마 동안 블록할지를 조정하는 것이다. 또한 poll()은 레코드 전체를 리턴하고, 레코드에는 토픽,파티션,파티션의 오프셋,키,값을 포함하고 있다. 한 번에 하나의 메시지만 가져오는 것이 아니라 여러개의 메시지를 가져오기 때문에 for-each로 데이터들을 처리하고 있다.



메시지의 순서


파티션이 3개인 토픽을 새로 만들어서 해당 토픽으로 메시지를 a,b,c,d,e 순서대로 보내보았다. 그리고 컨슈머 실행후 결과를 보니 a,d,b,e,c 순서대로 메시지가 들어왔다. 컨슈머에 문제가 있는 것인가? 아니면 진짜 내부적으로 문제가 있어서 순서가 보장되지 않은 것인가 궁금할 수 있다. 하지만 지극히 정상인 결과값이다.


먼저 해당 토픽에 메시지가 어떻게 저장되어 있는지 부터 확인해봐야 한다. 해당 토픽은 파티션이 3개로 구성되어 있기 때문에 각 파티션별로 메시지가 어떻게 저장되어 있는지 확인해봐야한다.



0번 파티션에는 b,e 1번 파티션에는 a,d 2번 파티션에는 c 데이터가 저장되어 있는 것을 볼수 있다. 즉, 컨슈머는 프로듀서가 어떤 순서대로 메시지를 보내는지 알수 없다. 단지 파티션의 오프셋 기준으로만 메시지를 가져올 뿐이다. 이말은 무엇이냐면, 카프카 컨슈머에서의 메시지 순서는 동일한 파티션내에서만 유지되고 파티션끼리의 메시지 순서는 보장하지 않는 것이다. 


카프카를 사용하면서 메시지의 순서를 보장해야 하는 경우에는 토픽의 파티션 수를 1로 설정한다. 하지만 단점도 존재한다. 메시지의 순서는 보장되지만 파티션 수가 하나이기 때문에 분산해서 처리할 수 없고 하나의 컨슈머에서만 처리할 수 있기 때문에 처리량이 높지 않다. 즉 처리량이 높은 카프카를 사용하지만 메시지의 순서를 보장해야 한다면 파티션 수를 하나로 만든 토픽을 사용해야 하며, 어느 정도 처리량이 떨어지는 부분은 감수해야한다.




컨슈머 그룹

카프카의 큰 장점 중 하나는, 하나의 토픽에 여러 컨슈머 그룹이 동시에 접속해 메시지를 가져 올 수 있다는 것이다. 이것은 기존의 다른 메시징큐 솔루션에서 컨슈머가 메시지를 가져가면 큐에서 삭제되어 다른 컨슈머가 가져갈 수 없다는 것과는 다른 방식인데 이 방식이 좋은 이유는 최근에 하나의 데이터를 다양한 용도로 사용하는 요구가 많아졌기 때문이다. 카프카가 이러한 기능제공이 가능한 이유는 파일시스템방식을 채택했기 이고, 컨슈머 그룹마다 각자의 오프셋을 별도로 관리하기 때문에 하나의 토픽에 두개의 컨슈머 그룹뿐만 아니라 더 많은 컨슈머 그룹이 연결되어도 다른 컨슈머 그룹에게 영향이 없이 메시지를 가져갈 수 있다. 이렇게 여러개의 컨슈머 그룹이 동시에 하나의 토픽에서 메시지를 가져갈 때는 컨슈머 그룹아이디를 서로 유일하게 설정해주어야한다.
 이전 이야기는 똑같은 데이터에 대해 요구가 다른 처리를 하기 위한 이야기 였다면, 이제 이야기 하려고하는 것은 하나의 토픽메시지를 여러개의 컨슈머가 나누어 처리하는 이야기이다.

만약 토픽에 a,b,c,d,e 라는 메시지가 들어왔고, 컨슈머는 한번에 하나의 메시지밖에 처리를 하지 못한다고 가정해보자. 그렇다면 컨슈머는 총 5번의 읽는 행위를 해야한다. 점차 메시지가 들어오는 양이 많아지면 해당 컨슈머의 처리 지연에 대한 영향은 점점 커질 것이다. 이러한 점을 해야결하기 위하여 하나의 컨슈머 그룹에 하나의 컨슈머가 아니라 여러 컨슈머를 그룹에 포함시켜 a,b,c,d,e 라는 메시지를 적절히 나누어 처리하게 하는 방법이다.

기본적으로 컨슈머 그룹 안에서 컨슈머들은 메시지를 가져오고 있는 토픽의 파티션에 대해 소유권을 공유한다. 컨슈머 그룹 내 컨슈머의 수가 보족해 프로듀서가 전송하는 메시지를 처리하지 못하는 경우에는 컨슈머를 추가해야하며, 추가 컨슈머를 동일한 컨슈머 그룹 내에 추가시키면 하나의 컨슈머가 가져오고 있던 메시지를 적절하게 나누어 가져오기 시작한다.만약 위의 그림처럼 만약 Consumer Group A가 C1만 있었고, C2라는 컨슈머를 동일한 그룹내에 추가했다면 P2,P3의 소유권이 C1에서 C2로 이동한다. 이것이 초반에 이야기 했던 리밸런스라고한다. 이렇게 리밸런스라는 기능을 통해 컨슈머를 쉽고 안전하게 추가할 수 있고 제거할 수도 있어 높은 가용성과 확장성을 확보할 수 있다. 하지만 이러한 리밸런스도 단점은 있다. 리밸런스를 하는 동안 일시적으로 컨슈머는 메시지를 가져올 수 없다. 그래서 리밸런스가 발생하면 컨슈머 그룹 전체가 일시적으로 사용할 수 없다는 단점이 있다.


위의 그림에서 Consumer Group B에 4개의 컨슈머가 있음에도 불구하고 처리가 계속 지연된다면 어떻게 될까? 이전처럼 해당 컨슈머 그룹에 컨슈머만 추가하면 될까? 아니다. 이뉴는 토픽의 파티션에는 하나의 컨슈머만 연결 할 수 있기 때문이다. 이말은 즉슨, 토픽의 파티션 수 만큼 최대 컨슈머 수가 연결될수 있는 것이다. 토픽의 파티션 수와 동일하게 컨슈머 수를 늘렸는데도 프로듀서가 보내는 메시지의 속도를 따라가지 못한다면 컨슈머만 추가하는 것이 아니라, 토픽의 파티션 수까지 늘려주고 컨슈머 수도 늘려줘야한다. 


이번에는 잘 동작하던 컨슈머 그룹 내에서 컨슈머 하나가 다운되는 경우를 생각해보자. 컨슈머가 컨슈머 그룹 안에서 멤버로 유지하고 할당된 파티션의 소유권을 유지하는 방법은 하트비트를 보내는 것이다. 반대로 생각해보면, 컨슈머가 일정한 주기로 하트비트를 보내다는 사실은 해당 파티션의 메시지를 잘 처리하고 있다는 것이다. 하트비트는 컨슈머가 poll 할때와 가져간 메시지의 오프셋을 커밋할 때 보내게 된다. 만약 컨슈머가 오랫동안 하트비트를 보내지 않으면 세션은 타임아웃되고, 해당 컨슈머가 다운되었다고 판단하여 리밸런스가 일어난다. 그 이후 다른 컨슈머가 다운된 컨슈머의 파티션의 할당 파티션을 맡게 되는 것이다.



커밋과 오프셋

컨슈머가 poll을 호출할 때마다 컨슈머 그룹은 카프카에 저장되어 있는 아직 읽지 않은 메시지를 가져온다. 이렇게 동작할 수 있는 것은 컨슈머 그룹이 메시지를 어디까지 가져갔는지 알 수 있기 때문이다. 컨슈머 그룹의 컨슈머들은 각각의 파티션에 자신이 가져간 메시지의 위치 정보(오프셋)을 기록한다. 각 파티션에 대해 현재 위치를 업데이트하는 동작을 커밋한다고 한다. 카프카는 각 컨슈머 그룹의 파티션별로 오프셋을 저장하기 위하여 __consumer_offsets 토픽을 만들고 오프셋을 저장한다. 리밸런스가 일어난 후 각각의 컨슈머는 이전에 처리했던 토픽의 다른 파티션을 할당 받게 되고 컨슈머는 새로운 파티션에 대해 가장 최근에 커밋된 오프셋을 일고 그 이후부터 메시지들을 가져오기 시작한다.


자동커밋

오프셋을 직접 관리해도 되지만, 각 파티션에 대한 오프셋 정보관리, 파티션 변경에 대한 관리 등이 매우 번거로울 수 있다. 그래서 카프카에서는 자동커밋 기능을 제공해준다. 자동 커밋을 사용하고 싶을 때는 컨슈머 옵션 중 enable.auto.commit=true로 설정하면 5초마다 컨슈머는 poll을 호출할 때 가장 마지막 오프셋을 커밋한다. 5초 주기는 기본 값이며, auto.commit.interval.ms 옵션을 통해 조정이 가능하다. 컨슈머는 poll을 요청할 때마다 커밋할 시간이 아닌지 체크하게 되고, poll 요청으로 가져온 마지막 오프셋을 커밋한다. 하지만 이 옵션을 사용할 경우 커밋 직전 리밸런스등의 작업이 일어나면 동일한 메시지에 대한 중복처리가 일어날 수 있다. 물론 자동커밋 주기를 작게 잡아 최대한 중복을 줄일 수 있지만 중복등을 완전하게 피할 수 는없다.


수동커밋

경우에 따라 자동 커밋이 아닌 수동 커밋을 사용해야하는 경우도 있다. 이러한 경우는 메시지 처리가 완료될 때까지 메시지를 가져온 것으로 간주되어서는 안되는 경우에 사용한다. 자동 커밋을 사용하는 경우라면 자동 커밋의 주기로 인해 일부 메시지들을 데이터베이스에는 자장하지 못한 상태로 컨슈머 장애가 발생한다면 해당 메시지들은 손실될 수도 있다. 이러한 경우를 방지하기 위해 컨슈머가 메시지를 가져오자마자 커밋을 하는 것이 아니라, 데이터베이스에 메시지를 저장한 후 커밋을 해서 위의 문제를 조금이나마 해결할 수 있다.(밑의 소스에서 그룹 아이디와 토픽등은 자신에 환경에 맞게 설정해주시길 바랍니다.)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import java.util.Arrays;
import java.util.Properties;
import java.time.Duration;
 
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
 
import lombok.extern.slf4j.Slf4j;
 
/**
 * Kafka Consumer Exam
 * @author yun-yeoseong
 *
 */
@Slf4j
public class KafkaBookConsumer1 {
    
    public static Properties init() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("group.id", "yoon-consumer");
        props.put("enable.auto.commit", "true");
        props.put("auto.offset.reset", "latest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }
    
    /**
     * 수동커밋
     */
    public static void commitConsume() {
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(init());){
            //토픽리스트를 매개변수로 준다. 구독신청을 한다.
            consumer.subscribe(Arrays.asList("yoon"));
            while (true) {
              //컨슈머는 토픽에 계속 폴링하고 있어야한다. 아니면 브로커는 컨슈머가 죽었다고 판단하고, 해당 파티션을 다른 컨슈머에게 넘긴다.
              ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
              for (ConsumerRecord<String, String> record : records)
                log.info("Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
              
              /*
               * DB로직(CRUD) 이후 커밋.
               */
              
              consumer.commitSync(); ->수동으로 커밋하여 메시지를 가져온 것으로 간주하는 시점을 자유롭게 조정할 수있다.
            }
        } finally {}
    }
        
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        commitConsume();
    }
 
}
 
cs



특정 파티션 할당

지금까지 컨슈머는 토픽을 subscribe하고, 카프카가 컨슈머 그룹의 컨슈머들에게 직접 파티션을 공정분배했다. 하지만 특별한 경우 특정 파티션에 대해 세밀하게 제어하길 원할 수도 있다. 이럴때에는 특정 컨슈머에게 특정 파티션을 직접 할당할 수 있다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
package com.kafka.exam;
 
import java.util.Arrays;
import java.util.Properties;
import java.time.Duration;
 
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
 
import lombok.extern.slf4j.Slf4j;
 
/**
 * Kafka Consumer Exam
 * @author yun-yeoseong
 *
 */
@Slf4j
public class KafkaBookConsumer1 {
    
    /**
     * 특정 파티션 할당
     */
    public static void specificPart() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        //특정 파티션을 할당하기 위해서는 기존과는 다른 그룹아이디 값을 주어야한다. 왜냐하면 컨슈머 그룹내에 같이 있게되면 서로의
        //오프셋을 공유하기 때문에 종료된 컨슈머의 파티션을 다른 컨슈머가 할당받아 메시지를 이어서 가져가게 되고, 오프셋을 커밋하게 된다.
        props.put("group.id", "specificPart");
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "latest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        String topicName = "yoon";
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(init());){
            //토픽리스트를 매개변수로 준다. 구독신청을 한다.
            TopicPartition partition0 = new TopicPartition(topicName, 0);
            TopicPartition partition1 = new TopicPartition(topicName, 1);
            consumer.assign(Arrays.asList(partition0,partition1));
            
            while (true) {
              //컨슈머는 토픽에 계속 폴링하고 있어야한다. 아니면 브로커는 컨슈머가 죽었다고 판단하고, 해당 파티션을 다른 컨슈머에게 넘긴다.
              ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
              for (ConsumerRecord<String, String> record : records)
                log.info("Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
              
              /*
               * DB로직(CRUD) 이후 커밋.
               */
              
              consumer.commitSync();
            }
        } finally {}
    }
    
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        specificPart();
    }
 
}
 
cs

특정 오프셋부터 메시지 가져오기


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
package com.kafka.exam;
 
import java.util.Arrays;
import java.util.Properties;
import java.time.Duration;
 
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
 
import lombok.extern.slf4j.Slf4j;
 
/**
 * Kafka Consumer Exam
 * @author yun-yeoseong
 *
 */
@Slf4j
public class KafkaBookConsumer1 {
    
    /**
     * 특정파티션에서 특정 오프셋의 메시지 가져오기
     */
    public static void specificOffset() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        //특정 파티션을 할당하기 위해서는 기존과는 다른 그룹아이디 값을 주어야한다. 왜냐하면 컨슈머 그룹내에 같이 있게되면 서로의
        //오프셋을 공유하기 때문에 종료된 컨슈머의 파티션을 다른 컨슈머가 할당받아 메시지를 이어서 가져가게 되고, 오프셋을 커밋하게 된다.
        props.put("group.id", "specificPartAndOffset");
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "latest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        String topicName = "yoon";
        
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(init());){
            //토픽리스트를 매개변수로 준다. 구독신청을 한다.
            TopicPartition partition0 = new TopicPartition(topicName, 0);
            TopicPartition partition1 = new TopicPartition(topicName, 1);
            consumer.assign(Arrays.asList(partition0,partition1));
            //0,1번 파티션의 2번 오프셋의 메시지를 가져와라
            consumer.seek(partition0, 2);
            consumer.seek(partition1, 2);
            while (true) {
              //컨슈머는 토픽에 계속 폴링하고 있어야한다. 아니면 브로커는 컨슈머가 죽었다고 판단하고, 해당 파티션을 다른 컨슈머에게 넘긴다.
              ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
              for (ConsumerRecord<String, String> record : records)
                log.info("Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
              
              /*
               * DB로직(CRUD) 이후 커밋.
               */
              
              consumer.commitSync();
            }
        } finally {}
    }
    
    public static void main(String[] args) {
        // TODO Auto-generated method stub
        specificOffset();
    }
 
}
 
cs


여기까지 자바소스로 다루어본 카프카 컨슈머 예제들이었습니다. 부족한 점이 많습니다. 혹시나 잘못된 점이 있다면 지적해주시면 감사하겠습니다!

저작자표시

'Middleware > Kafka&RabbitMQ' 카테고리의 다른 글

Kafka - Spring cloud stream kafka(스프링 클라우드 스트림 카프카)  (0) 2019.03.28
Kafka - Kafka Streams API(카프카 스트림즈) - 1  (2) 2019.03.26
Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI  (0) 2019.03.16
Kafka - Kafka(카프카) cluster(클러스터) 구성 및 간단한 CLI사용법  (0) 2019.03.13
Kafka - Kafka(카프카)의 동작 방식과 원리  (0) 2019.03.12
posted by 여성게
:
Middleware/Kafka&RabbitMQ 2019. 3. 16. 21:10

Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI

Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI



카프카 프로듀서란 메시지를 생산(produce)해서 카프카의 토픽으로 메시지를 보내는 역할을 하는 애플리케이션, 서버 등을 모두 프로듀서라고 부른다.

프로듀서의 주요 기능은 각각의 메시지를 토픽 파티션에 매핑하고 파티션의 리더에 요청을 보내는 것이다. 키 값을 정해 해당 키를 가진 모든 메시지를 동일한

파티션으로 전송할 수 있다. 만약 키 값을 입력하지 않으면, 파티션은 라운드 로빈(round-robin) 방식으로 파티션에 균등하게 분배된다.


이후의 모든 예제는 이전 포스팅에서 구성한 카프카 클러스터링 환경에서 진행하였습니다. 동일한 환경 구성을 

구축하고 예제를 진행하시려면 이전 포스팅을 참조하시길 부탁드립니다.


▶︎▶︎▶︎카프카 클러스터링




콘솔 프로듀서로 메시지를 보내 보기

우선 메시지를 보내기 위해서는 토픽이 카프카 내에 존재해야 한다. 만약 존재하지 않는 토픽으로 메시지를 보낼 경우에 자동으로 토픽이 
생성되는 옵션을 이용하려면 "auto.create.topics.enable = true" 옵션을 환경설정 파일에 넣어주면 된다. 
이번 예제는 직접 수동으로 토픽을 생성할 예정이다.

토픽생성은 카프카에 기본으로 제공하는 스크립트를 이용하여 생성한다.


위의 명령으로 test-topic이라는 이름으로 파티션이 1개로 구성되어 있고, 복제본이 자신을 포함한 3개인 토픽이 생성된다.

만약 복제본들이 어느 클러스터 인스턴스에 할당되어 있으며, 리더는 누구인지 그리고 ISR 그룹 구성등을 보고 싶다면 카프카에서 기본으로 제공하는
스크립트로 조회가능하다.


결과를 보면 Leader는 2번 클러스터 인스턴스이며, 파티션은 0번 한개 그리고 복제본이 1,2,3번의 클러스터 인스턴스에 생성되어 있고,

ISR그룹이 1,2,3 클러스터 인스턴스로 이루어져 있음을 알 수 있다. 여기서 ISR 그룹이란 복제본들이 들어있는 클러스터 인스턴스 그룹으로 볼 수 있다.

그리고 리더 선출의 후보의 전제조건이 클러스터가 ISR 그룹내에 존재하는 클러스터이여야만 리더 선출의 후보가 될 수 있다. ISR은 카프카가 도입한

하나의 개념이라고 볼 수 있다. 하나를 예제로 설명하면 현재 2번 클러스터가 리더인데, 만약 해당 리더가 다운되면 ISR그룹에서 2번 클러스터는 빠지게 되고

ISR그룹내 1,3 클러스터 중에서 카프카 컨트롤러가 리더를 선출하게 된다.




카프카 콘솔을 이용한 간단한 메시징 테스트이다.

사실 이런 예제는 클러스터 구축단계에서 다 해보았던 것이지만 이번 포스팅에서는 프로듀서의 몇몇 중요한 옵션들을 자바를 기반으로

설명할 것이다. 물론 spring cloud stream을 이용해 메시징 미들웨어에 의존적이지 않은 코드로도 작성가능 하지만 이번엔 조금더

로우레벨로 직접 자바코드를 이용해 카프카 시스템을 이용해볼 것이다.


자바를 이용한 프로듀서


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
/**
 * Kafka Producer Exam
 * @author yun-yeoseong
 *
 */
@Slf4j
public class KafkaBookProducer1 {
    
    public static Properties init() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return props;
    }
    
    /**
     * 메시지 전송의 성공/실패여부를 신경쓰지 않는다.
     * 카프카가 항상 살아있는 상태이고 프로듀서가 자동으로 재전송하기 때문에 대부분의 경우 성공적으로 전송되지만,
     * 일부 유실이 있을 수 있다.
     * 
     * send() method를 이용해 ProducerRecord를 보낸다. 메시지는 버퍼에 저장되고 별도의 스레드를 통해 브로커로 전송한다.(로그만 봐도 main이 아닌 별도의 스레드가 일을 처리)
     * send()는 자바 퓨처(Future) 객체로 RecordMetadata를 리턴받지만 리턴값을 무시하기 때문에 메시지가 성공적으로 전송되었는지 알수 없다.
     * 실 서비스 환경에서는 거의 사용하지 않는다.
     * 
     * 메시지를 보낸 후의 에러는 무시하지만, 전송전의 에러는 잡아서 처리할 수 있다.
     */
    public static void sendNoConfirmResult() {
        long start = System.currentTimeMillis();
        try(Producer<String, String> producer = new KafkaProducer<>(init())){
            producer.send(new ProducerRecord<String, String>("yeoseong", "Apache Kafka is a distributed streaming platform-sendNoConfirmResult()"));
        }catch (Exception e) {}
        long end = System.currentTimeMillis();
        log.info("sendNoConfirmResult - during time : "+ (end-start));
    }
    
    /**
     * 리턴 값으로 Future를 반환하는데, get()을 호출하여 결과를 받아 메시지가 성공적으로 전송됬는지 체크한다. 
     * 메인 스레드가 block되는 상황이 있지만, 신뢰성있는 메시지 전송을 보장한다.
     */
    public static void sendSync() {
        long start = System.currentTimeMillis();
        try(Producer<String, String> producer = new KafkaProducer<>(init())){
            RecordMetadata meta = producer.send(new ProducerRecord<String, String>("yeoseong", "Apache Kafka is a distributed streaming platform-sendSync()")).get();
            log.info("Partition: {}, Offset: {}",meta.partition(),meta.offset());
        }catch (Exception e) {}
        long end = System.currentTimeMillis();
        log.info("sendSync - during time : "+ (end-start));
    }
    
    /**
     * 콜백방식으로 비동기 전송을 한다. 메시지 전송이 완료 혹은 실패되면,
     * 브로커쪽에서 콜백으로 onCompletion을 호출한다. 만약 실패하면 Exception 객체가 담긴다.
     * org.apache.kafka.clients.producer.Callback
     * 
     * 로그를 보면 main이 아닌 별도의 카프카 스레드에서 콜백을 호출한다.
     */
    public static void sendAsync() {
        long start = System.currentTimeMillis();
        try(Producer<String, String> producer = new KafkaProducer<>(init())){
            producer.send(new ProducerRecord<String, String>("yeoseong", "Apache Kafka is a distributed streaming platform-sendAsync()"),new KafkaCallback());
        }catch (Exception e) {}
        long end = System.currentTimeMillis();
        log.info("sendAsync() - during time : "+ (end-start));
    }
    
    /**
     * 프로듀서에서 키값까지 같이 메시지를 보내면 키에 해당하는 파티션에만 메시지가 들어간다.
     * 하지만 키를 포함시키지 않는다면 라운드 로빈 방식으로 파티션마다 균등분배된다.
     * 
     * evenkey&oddkey는 각각 동일한 파티션으로만 데이터를 전송하게 된다.
     */
    public static void sendWithKey() {
        long start = System.currentTimeMillis();
        String topicName = "yoon";
        String oddKey="1";
        String evenKey="2";
        
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "1");
        props.put("compression.type", "gzip");
        
        try(Producer<String, String> producer = new KafkaProducer<>(props)){
            
            IntStream.rangeClosed(1, 10).forEach(i->{
                if(i%2==1) {
                    producer.send(new ProducerRecord<String, String>(topicName, oddKey ,i+" - Apache Kafka is a distributed streaming platform-sendWithKey() oddKey "+oddKey));
                }else {
                    producer.send(new ProducerRecord<String, String>(topicName, evenKey ,i+" - Apache Kafka is a distributed streaming platform-sendWithKey() evenKey "+evenKey));
                }
            });
            
            
        }catch (Exception e) {}
        long end = System.currentTimeMillis();
        log.info("sendWithKey() - during time : "+ (end-start));
    }
    
    
    public static void main(String[] args) {
        
//        sendNoConfirmResult();
//        sendSync();
//        sendAsync();
        sendWithKey();
    }
 
}
 
@Slf4j
class KafkaCallback implements Callback{
 
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        // TODO Auto-generated method stub
        if(!ObjectUtils.isEmpty(metadata)) {
            log.info("Partition: {}, Offset: {}",metadata.partition(),metadata.offset());
        }else {
            log.error("KafkaCallback - Exception");
        }
    }
    
}
 
cs

코드는 몇개의 예제를 메소드 형태로 구분하여 만들었다.

첫번째 메소드는 프로듀서에서 서버로 메시지를 보내고 난 후에 성공적으로 도착했는지까지 확인하지 않는다. 카프카가 항상 살아있는 상태이고
프로듀서가 자동으로 재전송하기 때문에 대부분의 경우 성공적으로 전송되지만, 일부 메시지는 손실이 있을 수도 있다. send()는 Future 객체로 RecordMetadata를 리턴받지만 그 리턴값을 객체로 받지 않고 있기 때문에 성공적으로 메시지가 전달되었는지 알 수 없다. 대부분 테스트 용도로만 사용하고 실서비스에서는
이용하지 않는 방법이다.

두번째 메소드는 동기 전송방법이다. 프로듀서는 메시지를 보내고 Future의 get()를 이용하여 Future를 기다린 후에 send()가 성공했는지 실패했는지 확인한다.
Future의 get()은 해당 작업 스레드를 블럭한 상태에서 Future의 리턴을 기다리기 때문에 동기 전송방법인 것이다.

세번째 메소드는 비동기 전송방법이다. 프로듀서는 send() 메소드를 콜백과 같이 호출하고 카프카 브로커에서 응답을 받으면 콜백한다. 
비동기적으로 전송한다면 응답을 기다리지 않기 때문에 더욱 빠른 전송이 가능하다. 콜백을 사용하기 위해 org.apache.kafka.clients.producer.Callback을
구현하는 클래스가 필요하다.(사실 람다식을 이용해서 매개변수로 넘겨주면 클래스는 따로 구현하지 않아도된다.) 카프카가 오류를 리턴하면 onCompletion()는 예외를 갖게 된다.

마지막 메소드는 키값을 추가하여 메시지를 보낼 경우이다. 키값을 포함하였을 경우에 컨슈머에는 어떻게 메시지가 전달 될지 콘솔 컨슈머를
이용하여 결과를 확인해볼 것이다.

우선 파티션 2개짜리인 토픽을 생성한다.


마지막 메소드를 main 메소드에서 실행한 후에 각각의 파티션으로 컨슈머를 실행했을 경우 결과가 어떻게 나오지는 확인한다.



2번 키를 포함시킨 메시지는 모두 0번 파티션으로 갔고, 1번 키를 포함시킨 메시지는 모두 1번 파티션으로 메시지가 전송되었다.

물론 특정파티션이 아닌 토픽으로 컨슈머를 실행시켰다면 모든 메시지 10개가 결과에 포함되었을 것이다. 이렇게 키값을 이용하여

특정 파티션으로만 메시지를 보낼 수 있는 기능을 제공하므로 다양하게 활용 가능하다.




프로듀서 주요옵션

Producer 주요 옵션

1)bootstrap.servers localhost:9092,localhost:9093,localhost:9094

:카프카 클러스터에 클라이언트가 처음 연결을 위한 호스트와 포트정보로 구성된 리스트 정보를 나타낸다. 전체 카프카 리스트가 아닌 호스트 하나만 입력해 사용할 수 있지만

카프카 인스턴스가 죽으면 그 클라이언트는 더이상 접속 할 수 없습니다. 즉, 전체 클러스터 인스턴스 목록을 옵션으로 넣는 것이 좋습니다.


2)acks 0,1,all , —request-required-acks 0,1,all

-acks = 0 

0으로 설정하면 제작자는 서버로부터의 승인을 전혀 기다리지 않습니다. 레코드가 소켓 버퍼에 즉시 추가되고 전송 된 것으로 간주됩니다. 이 경우 서버가 레코드를 수신했음을 보증 할 수 없으며 재시도 구성이 적용되지 않습니다 (클라이언트는 일반적으로 어떤 실패도 알 수 없으므로). 각 레코드에 대해 다시 주어진 오프셋은 항상 -1로 설정됩니다.


-acks = 1 

만약 1로 설정하는 경우 리더는 데이터를 기록하지만, 모든 팔로워는 확인하지 않습니다. 이 경우 일부 데이터 손실이 있을수 있습니다.


-acks = all 

acks=-1과 동일하며, 리더는 ISR의 팔로워로부터 데이터에 대한 ack를 기다립니다. 하나의 팔로워가 있는 한 데이터는 손실되지 않으며, 데이터 무손실에 대해 가장 강력하게 보장합니다.


3)buffer.memory

프로듀서가 카프카 서버로 데이터를 보내기 위해 잠시 대기(배치 전송이나 딜레이 등)할 수 있는 전체 메모리 바이트입니다.


4)compression.type

프로듀서가 데이터를 압축해서 보낼 수 있는데, 어떤 타입으로 압축할지를 정할 수 있다. 옵션으로 none,gzip,snappy,lz4 같은 다양한 포맷이 있다.


5)retries

일시적인 오류로 인해 전송에 실패한 데이터를 다시 보내는 횟수


6)batch.size

프로듀서는 같은 파티션으로 보내는 여러 데이터를 함께 배치로 보내려고 시도한다. 이러한 동작은 클라이언트와 서버 양쪽에 성능적인 측면에서 도움이 된다. 이 설정으로 배치 크기 바이트 단위를 조정할 수 있다. 정의된 크기보다 큰 데이터는 배치를 시도하지 않는다. 배치를 보내기전 클라이언트 장애가 발생하면 배치 내에 있던 메시지는 전달되지 않는다. 만약 고가용성이 필요한 메시지의 경우라면 배치 사이즈를 주지 않는 것도 하나의 방법이다.


7)linger.ms

배치형태의 메시지를 보내기 전에 추가적인 메시지들을 위해 기다리는 시간을 조정한다. 카프카 프로듀서는 지정된 배치 사이즈에 도달하면 이 옵션과 관계없이 즉시 메시지를 전송하고, 배치 사이즈에 도달하지 못한 상황에서 linger.ms 제한 시간에 도달했을때 메시지들을 전송한다. 0이 기본값(지연없음)이며, 0보다 큰 값을 설정하면 지연 시간은 조금 발생하지만 처리량이 좋아진다.


8)max.request.size

프로듀서가 보낼 수 있는 최대 메시지 바이트 사이즈. 기본값은 1MB이다.


나머지 Kerberos, ssl과 타임관련등의 설정들은 https://kafka.apache.org/documentation/#producerconfigs을 참고



server.propertis ->acks=all 에 따른 server.properties 설정정보 변경

만약 밑의 설정이 없다면 설정파일의 맨밑에 작성

min.insync.replicas=n

이 설정은 acks=all로 했을 경우 리더를 포함한 인스턴스 n개의 승인(데이터복제,저장)을 받으면 바로 프로듀서에게 결과를 전달한다.

최적은 acks=all, min.insync.replicas=2, replication factor=3 ->카프카문서에 명시되어있음.



메시지 전송 방법

프로듀서의 옵션 중 acks 옵션을 어떻게 설정하는지에 따라 카프카로 메시지를 전송할 때 메시지 손실 여부와 메시지 전송 속도 및 처리량이 달라진다.

1) 메시지 손실 가능성이 높지만 빠른 전송이 필요한 경우(acks=0 으로 설정)
메시지를 전송할 때 프로듀서는 카프카 서버에서 응답을 기다리지 않고, 메시지를 보낼 준비가 되는 즉시 다음 요청을 보낸다. 하지만 이런 방법을
이용한다면 자신이 보낸 메시지에 대해 결과를 기다리지 않기때문에 메시지 유실이 있을 수도 있다.


2) 메시지 손실 가능성이 적고 적당한 속도의 전송이 필요한 경우(acks=1 으로 설정)

이 옵션은 앞선 옵션과 달리 프로듀서가 카프카로 메시지를 보낸 후 보낸 메시지에 대해 메시지를 받는 토픽의 리더가 잘 받았는지 확인(acks)한다.

응답 대기 시간 없이 계속 메시지만 보내던 방법과 달리 확인을 하는 시간이 추가되어 메시지를 보내는 속도는 약간 떨어지게 되지만

메시지의 유실 가능성이 조금 줄어든다. 하지만 메시지를 받는 리더가 장애가 발생하면 메시지 유실 가능성이 있다. 하지만 보통 프로듀서 애플리케이션으로

많이 사용하는 로그스태시(logstash), 파일비트(Filebeat) 등에서는 프로듀서의 acks 옵션을 기본 1로 하고 있다. 특별한 경우가 아니라면

속도와 안전성을 확보할 수 있는 acks=1을 사용하는 것을 추천한다. 물론 메시지 유실에 큰 영향이 없는 서비스라는 조건하이다.


3) 메시지 전송 속도는 느리지만 메시지 손실이 없어야 하는 경우(acks=all 으로 설정)

acks=all의 동작 방법은 프로듀서가 메시지를 전송하고 난후 리더가 메시지를 받았는지 확인하고 추가로 팔로워까지 메시지를 잘받았는지(복제) 확인 하는 것이다.

속도적인 측면으로 볼때, acks 옵션 중에서 가장 느리지만 메시지 손실을 허용하지 않은 경우 사용하는 방법이다. acks=all을 완벽하게 사용하고자 한다면

프로듀서의 설정 뿐 아니라 브로커의 설정도 같이 조정해야한다. 브로커의 설정에 따라 응답 확인을 기다리는 수가 달라지게 된다.


-프로듀서의 acks=all과 브로커의 min.insync.replicas=1

이것은 acks=all 설정을 하였고 replication-factor가 예를 들어 3개라고 해도, 응답을 하나의 클러스터에서만 받게 된다.

즉, acks=1과 동일한 설정이 된다.


-프로듀서의 acks=all과 브로커의 min.insync.replicas=2

프로듀서가 메시지를 보내면 2개의 클러스터에 응답을 기다리게 된다.


-프로듀서의 acks=all과 브로커의 min.insync.replicas=3

만약 리플리케이션 팩터가 3이라면 위의 설정은 모든 클러스터 인스턴스의 응답 결과를 기다린다. 여기서 하나의 클러스터가 장애가

발생한다면 프로듀서는 메시지를 보낼 수 없게 될 것이다.


위에서 설명하였지만 카프카가 공식적으로 권장하는 옵션은

acks=all, min.insync.replicas=2, replication factor=3이다.


저작자표시

'Middleware > Kafka&RabbitMQ' 카테고리의 다른 글

Kafka - Kafka Streams API(카프카 스트림즈) - 1  (2) 2019.03.26
Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI  (2) 2019.03.24
Kafka - Kafka(카프카) cluster(클러스터) 구성 및 간단한 CLI사용법  (0) 2019.03.13
Kafka - Kafka(카프카)의 동작 방식과 원리  (0) 2019.03.12
Springboot - Rabbitmq를 이용한 비동기 메시징 서비스(리액티브 마이크로서비스)  (2) 2019.02.18
posted by 여성게
:
Middleware/Kafka&RabbitMQ 2019. 3. 13. 10:38

Kafka - Kafka(카프카) cluster(클러스터) 구성 및 간단한 CLI사용법

Kafka - Kafka(카프카) cluster(클러스터) 구성 및 간단한 CLI사용법




▶︎▶︎▶︎카프카란?


이전 포스팅에서는 메시징 시스템은 무엇이고, 카프카는 무엇이며 그리고 카프카의 특징과 다른 메시지 서버와의 차이점에 대한

포스티이었습니다. 이번 포스팅은 간단하게 카프카3대를 클러스터링 구성을 하여 서버를 띄우고 CLI를 이용하여 간단히

카프카를 사용해보려고 합니다. 카프카는 중앙에서 많은 서비스 시스템의 데이터를 받아서 다른 시스템으로 받아주는 역할을 하는

메시지 시스템으로 MSA에서는 없어선 안되는 존재가 되었습니다. 그렇다면 이렇게 중요한 카프카를 한대만 띄워서 프로덕트 환경에서

운영한다는 것은 과연 안전한 생각일까요? 아닙니다. 여러대를 클러스터링 구성하여 고가용성을 높혀야 운영환경에서도 안전하고 신뢰성있는

메시지 시스템 구성이 될것입니다. 






위의 그림은 카프카를 여러대 클러스터링을 했을 경우의 구성도 입니다. 일단 중요한 것은 Zookeeper라는 존재입니다. 주키퍼는

이전 포스팅에서 Zookeeper란 무엇인가? 그리고 Solr 클러스터링 구성을 했을 때 다뤘던 내용이므로 이번 포스팅에서는 생략합니다.




▶︎▶︎▶︎Zookeeper란?

▶︎▶︎▶︎Solr Cluster 구성


간단하게 이야기하면 주키퍼란 분산환경 애플리케이션을 중앙에서 관리해주는 디렉토리 구조의 분산 코디네이터라고 보시면 됩니다. 지금부터는

실습에 초점을 마추겠습니다.



우선은 주키퍼를 설치해줍니다. 이번 실습의 버전은 3.4.12 버전기준입니다.

설정파일을 만지기 전에 주키퍼를 위한 디렉토리 구성을 진행하겠습니다. 저는 카프카 실습을 위해 별도의 하나의 디렉토리를 구성하였고,

그안에 주키퍼와 카프카에 대한 파일들을 모두 넣어놓았습니다. 우선 다음 예제는 주키퍼와 카프카를 별도의 디렉토리에 설치했다는 가정하에 진행합니다.

주키퍼는 링크를 참조하셔서 총 3대로 띄워주시면 됩니다.(Master-slave 관계의 클러스터)


카프카도 동일한 디렉토리에 설치해줍니다. 편의상 카프카의 디렉토리 루트를 $KAFKA로 지칭합니다.


우선 진행하기 앞서 클러스터 구성에서 인스턴스 각각이 자신의 메타 데이터와 카프카가 데이터를 쓸 디렉토리를 만들어줍니다.

kdata1,kdata2,kdata3 디렉토리를 만들어줍니다. 그리고 $KAFKA/config 밑에 있는 server.properties를 2장 복사해줍니다.

그리고 해당 파일 안에 설정들을 클러스터 구성에 가장 기본이 되는 설정으로 바꿔줍니다.


broker.id = 1,broker.id = 2,broker.id = 3 으로 총 3개의 serverN.properties에 작성해줍니다. 이것은

주키퍼 myid와 비슷한 역할을 하는 클러스터 인스턴스 구분용 ID값입니다.

그리고 지금 실습은 로컬에 3대의 카프카서버를 띄워서 클러스터링 구성을 할것임으로 각각 서버의 포트를 달리 지정해줍니다.


listeners=PLAINTEXT://:9092,listeners=PLAINTEXT://:9093,listeners=PLAINTEXT://:9094 으로 각각 파일에 포트를 구분해줍니다.


log.dirs=$KAFAKA/kdata1,log.dirs=$KAFAKA/kdata2,log.dirs=$KAFAKA/kdata3 으로 각각 카프카 서버가 자신의

메타 데이터를 쓸 디렉토리를 구분지어줍니다. 필자는 처음에 하나의 디렉토리만 생성해서 공유했더니 충돌 문제가 있어

각각 별도의 디렉토리 구성을 가져갔습니다.



마지막으로 주키퍼와의 연동 설정입니다.

zookeeper.connect=localhost:2181,localhost:2182,localhost:2183/kafka-broker 으로 각 설정 파일에 동일하게 작성해줍니다.

여기에서 조금 특이한것이 ~/kafka-broker 입니다. 이것은 무엇이냐면 만약 각각 다른 용도의 카프카 클러스터를 2세트 운영한다고 가정하고,

주키퍼는 같은 주키퍼 클러스터를 이용한다고 합니다. 만약 ~/kafka-broker를 입력하지 않으면 카프카 클러스터 2세트가 주키퍼의 루트 디렉토리에 동일한

구조의 데이터를 쓰게 되고 그러면 카프카 클러스터 2세트가 데이터 충돌이 나게됩니다. 그렇기 때문에 ~/kafka-broker라고 입력하여 주키퍼의 루트디렉토리가

아닌 /kafka-broker/.. 라는 디렉토리를 하나 새로 생성하여 아예 카프카 클러스터 1대의 전용 디렉토리를 구성해주는 겁니다. 여기까지 진짜 기본적인

설정으로 클러스터 설정을 마쳤습니다. 사실은 이정도 설정으로는 운영환경에서는 운영하기 힘들것입니다. 나머지 설정관련해서는 카프카 공식 홈페이지에

영어로 친절하게? 작성되어 있으니 참고 부탁드립니다.






이제 $KAFKA/bin/kafka-server-start.sh -daemon ../config/server.properties 명령으로 3개의 카프카를 실행시켜줍니다. 물론 선행조건은

주키퍼가 실행되었다는 조건입니다. 만약 지금까지 잘 따라오셨다면 문제없이 실행됩니다. 마지막으로 간단한 CLI를 이용한 producer,consumer 예제입니다.



위의 명령으로 큐의 역할로 사용될 토픽을 생성해줍니다. 만약 토픽을 잘못 생성하신 분들을 위해서 토픽을 삭제하는 명령입니다.



Noting 메시지와 함께 토픽이 삭제되었습니다. 



프로듀서를 실행시켜줍니다. 설정 인자들은 설명없어도 직관적으로 보입니다.



메시지를 보냅니다.



또 한번 보냅니다.



컨슈머 쪽을 확인해보면 메시지가 잘와있는 것을 확인할 수 있습니다. 여기까지 카프카 클러스터 구성과 간단한 CLI를 이용하여 카프카를 사용해보았습니다.

저작자표시

'Middleware > Kafka&RabbitMQ' 카테고리의 다른 글

Kafka - Kafka Streams API(카프카 스트림즈) - 1  (2) 2019.03.26
Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI  (2) 2019.03.24
Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI  (0) 2019.03.16
Kafka - Kafka(카프카)의 동작 방식과 원리  (0) 2019.03.12
Springboot - Rabbitmq를 이용한 비동기 메시징 서비스(리액티브 마이크로서비스)  (2) 2019.02.18
posted by 여성게
:
Middleware/Kafka&RabbitMQ 2019. 3. 12. 22:30

Kafka - Kafka(카프카)의 동작 방식과 원리

Kafka - Kafka(카프카)의 동작 방식과 원리

 

 

Kafka는 기본적으로 메시징 서버로 동작합니다. 여기서 메시징 시스템에 대해 간단히 살펴보자면 

메시지라고 불리는 데이터 단위를 보내는 측(publisher,producer)에서 카프카에 토픽이라는 각각의 메시지 저장소에 데이터를 저장하면, 

가져가는 측(subscriber, consumer)이 원하는 토픽에서 데이터를 가져가게 되어 있습니다. 즉, 메시지 시스템은 중앙에 메시징 시스템 서버를

두고 이렇게 메시지를 보내고(publish) 받는(subscriber) 형태의 통신 형태인 pub/sub 모델의 통신구조입니다.

 

여기서 미담이지만, 카프카의 창시자인 제이 크렙스는 대학 시절 문학 수업을 들으며 소설가 프란츠 카프카에 심취했습니다. 자신의 팀이 새로 개발할 시스템이

데이터 저장과 기록, 즉 쓰기에 최적화된 시스템이었기에 이런 시스템은 글을 쓰는 작가의 이름을 사용하는 것이 좋겠다 생각하여 

카프카라는 이름으로 애플리케이션이 탄생하게 되었습니다.

 

 

 

 

pub/sub은 비동기 메시징 전송 방식으로서, 발신자의 메시지에는 수신자가 정해져 있지 않은 상태로 발행합니다. 구독을 신청한 수신자만이 정해진

메시지를 받을 수 있습니다. 또한 수신자는 발신자 정보가 없어도 원하는 메시지만을 수신가능합니다.

 

 

펍/섭 구조가 아닌 일반적인 통신은

 

A ---------------> B

C ---------------> D

 

형태로 통신을 하게 됩니다. 이럴 경우 빠른 전송 속도와 전송 결과를 신속하게 알 수 있는 장점이 있지만, 장애가 발생한 경우

적절한 처리를 수신자 측에서 해주지 않는다면 문제가 될 수 있습니다. 그리고 통신에 참여하는 개체수가 많아 질 수록 

일일이 다 연결하고 데이터를 전송해야 하기 때문에 확장성이 떨어집니다.

 

하지만 펍/섭구조는 

 

         ==========

A ---------------->=메시지서버= ---------->B

C ---------------->========== ---------->D

 

프로듀서가 메시지를 컨슈머에게 직접 전달하는 방식이 아니라 중간의 메시징 시스템에 전달합니다. 이때 메시지 데이터와 수신처 ID를

포함시킵니다. 메시징 시스템의 교환기가 메시지의 수신처ID 값을 확인한 다음 컨슈머들의 큐에 전달합니다. 컨슈머는 자신들의

큐를 모니터링하고 있다가, 큐에 메시지가 전달되면 이 값을 가져갑니다. 이렇게 구성할 경우 장점은 혹시나 통신 개체가 하나 빠지거나 수신 불가능 상태가 되었을 

때에도, 메시징 시스템만 살아 있다면 프로듀서에서 전달된 메시지는 유실되지 않고, 장애에서 복구한 통신 개체가 살아나면 다시 메시지를

안전하게 전달할 수 있습니다. 그리고 메시징 시스템을 중심으로 연결되기 때문에 확장성이 용이합니다. 물론 직접 통신 하지 않기 때문에

메시지가 정확하게 전달되었는지 확인하는 등의 코드가 들어가 복잡해지고, 메시징 서버를 한번 거쳐서 가기에 통신속도가 느린 단점도 있습니다.

하지만 메시징 시스템이 주는 장점때문에 이러한 단점은 상쇄될만큼 크지 않다고 생각합니다.(비동기가 주는 장점 등)

 

(밑의 설명을 두고 넣은 그림은 아님)

 

 

기존 메시징 시스템을 사용하는 펍/섭 모델은 대규모 데이터(데이터 단건당 수MB 이상)를 메시징 시스템을 통해 전달하기 보다는 간단한 이벤트(수KB정도)를 서로 전송하는데 주로 사용되었습니다. 왜냐하면 메시징 시스템 내부의 교환기의 부하, 각 컨슈머들의 큐관리, 큐에 전달되고 가져가는 메시지의 정합성, 전달 결과를

정확하게 관리하기 위한 내부 프로세스가 아주 다양하고 복잡했기 때문입니다.

즉, 기존 메시징 시스템은 속도와 용량보단 메시지의 보관,교환,전달 과정에서의 신뢰성을 보장하는 것에 중점을 두었습니다.

 

카프카는 메시징 시스템이 지닌 성능의 단점을 극복하기 위해, 메시지 교환 전달의 신뢰성 관리를 프로듀서와 컨슈머 쪽으로 넘기고, 부하가 많이 걸리는 교환기 기능

역시 컨슈머가 만들 수 있게 함으로써 메시징 시스템 내에서의 적업량을 줄이고 이렇게 절약한 작업량을 메시지 전달 성능에 집중시켜 

고성능 메시징 시스템을 만들어 냈습니다.

 

 

 

 

 

 

카프카의 특징

프로듀서와 컨슈머의 분리 - 카프카는 메시징 전송 방식 중 메시지를 보내는 역할과 받는 역할이 완벽하게 분리된 펍/섭 방식을 적용했습니다. 각 서비스서버들은 다른 시스템의 상태 유무와 관계없이 카프카로 메시지를 보내는 역할만 하면 되고, 마찬가지로 메시지를 받는 시스템도서비스 서버들의 상태유무와 관계없이 카프카에 저장되어 있는 메시지만 가져오면 됩니다.
멀티 프로듀서, 멀티 컨슈머 - 카프카는 하나의 토픽에 여러 프로듀서 또는 컨슈머들이 접근 가능한 구조로 되어있습니다. 하나의 프로듀서가 하나의토픽에만 메시지를 보내는 것이 아니라, 하나 또는 하나 이상의 토픽으로 메시지를 보낼 수 있습니다. 컨슈머는 역시 하나의 토픽에서만메시지를 가져오는 것이 아니라, 하나 또는 하나 이상의 토픽으로부터 메시지를 가져올 수 있습니다. 이렇게 멀티 프로듀서와 멀티 컨슈머를구성할 수 있기 때문에 카프카는 중앙 집중형 구조로 구성할 수 있게 됩니다.
디스크에 메시지 저장 - 카프카가 기존의 메시징 시스템과 가장 다른 특징 중 하나는 메모리에 데이터를 쓰는 것이 아니라, 바로 디스크에 메시지를 저장하고유지하는 것입니다. 일반적인 메시징 시스템들은 컨슈머가 메시지를 읽어가면 큐에서 바로 삭제하는데, 카프카는 컨슈머가 데이터를 가져가도 일정 기간동안데이터를 유지합니다. 트래픽이 급증해도 컨슈머는 디스크에 안전하게 저장된 데이터를 손실 없이 가져갈 수 있습니다. 또한 앞서 멀티 컨슈머가 가능한 것도이렇게 디스크에 데이터를 쓰기 때문입니다. 디스크에 데이터를 쓰기 때문에 처리가 느릴 것이라는 생각을 할 수 있기도한대, 기존의 디스크 사용법과는다르게 디스크를 나눠서 쓰는 것이 아니라, 특정 영역의 디스크에 순차적으로 쓰기 때문에 읽어가는 디스크의 영역의 범위를 확 줄일 수 있어 읽어가는 속도는조금은 빨라집니다. 그리고 카프카는 무조건 데이터를 디스크에 쓴다기 보다는 VM메모리의 일부를 페이지 캐싱으로 사용하기 때문에 속도가 빠릅니다.
확장성 - 카프카는 확장이 매우 용이하도록 설계되어 있습니다. 하나의 카프카 클러스터는 3대부터 수십대의 브로커로도 메시지 서버의 중지없이 확장 가능합니다.
높은성능 - 카프카는 내부적으로 고성능을 유지하기 위해 분산 처리, 배치 처리 등 다양한 기법을 사용하고 있습니다.(병렬 처리를 위한 토픽 파티셔닝 기능)
이러한 카프카의 사용으로 연동되는 애플리케이션들의 느슨한 결합도를 유지할 수 있게 됩니다. 이번 포스팅은 간단한 카프카의 소개였고,다음 포스팅부터는 진짜 카프카를 이용한 예제 포스팅이 될 것 같습니다.

 

저작자표시

'Middleware > Kafka&RabbitMQ' 카테고리의 다른 글

Kafka - Kafka Streams API(카프카 스트림즈) - 1  (2) 2019.03.26
Kafka - Kafka Consumer(카프카 컨슈머) Java&CLI  (2) 2019.03.24
Kafka - Kafka Producer(카프카 프로듀서) In Java&CLI  (0) 2019.03.16
Kafka - Kafka(카프카) cluster(클러스터) 구성 및 간단한 CLI사용법  (0) 2019.03.13
Springboot - Rabbitmq를 이용한 비동기 메시징 서비스(리액티브 마이크로서비스)  (2) 2019.02.18
posted by 여성게
:
Middleware/Redis 2019. 3. 1. 14:02

Springboot,redis - Redis repository 간단한사용법!

Springboot,redis - Redis repository 간단한사용법!



우선 처음에 조금 헤메긴 했지만, Redis Repository를 사용하려면 기존에 어떠한 Datasource라도 존재를 해야하는 것 같다. 그래서 임의로 인메모리디비인 H2를 dependency하였고, 모든 Datasource는 기본설정으로 두었다. 이렇게 인메모리 디비 데이터소스를 이용함에도 불구하고, 실제로는 Redis에 데이터가 삽입되는 것을 볼수 있다. 이유는 잘모르지만....아시는 분 있으시면 알려주세요...




1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.3.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.brownfield.pss</groupId>
    <artifactId>redis-cluster</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>redis-cluster</name>
    <description>Demo project for Spring Boot</description>
 
    <properties>
        <java.version>1.8</java.version>
    </properties>
 
    <dependencies>
        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.4</version>
            <scope>provided</scope>
        </dependency>
        <!-- H2 -->
        <dependency>
            <groupId>com.h2database</groupId>
            <artifactId>h2</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.json/json -->
        <dependency>
            <groupId>org.json</groupId>
            <artifactId>json</artifactId>
            <version>20160810</version>
        </dependency>
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
 
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
 
</project>
 
cs


pom.xml이다. 일단 Datasource가 필요하다라는 예외메시지에 H2 인메모리 디비를 dependency했다.


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
/**
 * Redis Configuration
 * @author yun-yeoseong
 *
 */
@Configuration
@EnableRedisRepositories
public class RedisConfig {
    
    
    /**
     * Redis Cluster 구성 설정
     */
    @Autowired
    private RedisClusterConfigurationProperties clusterProperties;
    
    /**
     * JedisPool관련 설정
     * @return
     */
    @Bean
    public JedisPoolConfig jedisPoolConfig() {
        return new JedisPoolConfig();
    }
    
    
    /**
     * Redis Cluster 구성 설정
     */
    @Bean
    public RedisConnectionFactory jedisConnectionFactory(JedisPoolConfig jedisPoolConfig) {
        return new JedisConnectionFactory(new RedisClusterConfiguration(clusterProperties.getNodes()),jedisPoolConfig);
    }
    
    /**
     * RedisTemplate관련 설정
     * 
     * -Thread-safety Bean
     * @param jedisConnectionConfig - RedisTemplate에 설정할 JedisConnectionConfig
     * @return
     */
    @Bean(name="redisTemplate")
    public RedisTemplate redisTemplateConfig(JedisConnectionFactory jedisConnectionConfig) {
        
        RedisTemplate redisTemplate = new RedisTemplate<>();
 
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        redisTemplate.setConnectionFactory(jedisConnectionConfig);
        
        return redisTemplate;
        
    }
//    @Bean
//    RedisTemplate<?, ?> redisTemplate(RedisConnectionFactory connectionFactory) {
//    
//      RedisTemplate<byte[], byte[]> template = new RedisTemplate<>();
//      template.setConnectionFactory(connectionFactory);
//      return template;
//    }
    
    
}
 
cs


Redis Config이다. @EnableRedisRepositories 어노테이션을 이용하여 레디스 레포지토리를 이용한다고 명시하였다. 나머지 설정은

이전 포스팅에서 구성했던 Redis Cluster 환경과 동일하게 진행하였다.
1
2
3
4
5
6
7
8
package com.spring.redis;
 
import org.springframework.data.repository.CrudRepository;
 
public interface RedisRepository extends CrudRepository<RedisEntity, Long> {
    public RedisEntity findByFirstname(String firstname);
}
 
cs

RedisRepositry 인터페이스이다. 사용법은 JPA 레포지토리랑 동일한것 같다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
package com.spring.redis;
 
import java.io.Serializable;
 
import org.springframework.data.annotation.Id;
import org.springframework.data.redis.core.RedisHash;
import org.springframework.data.redis.core.index.Indexed;
 
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
 
@RedisHash("person")
@Getter
@Setter
@ToString
public class RedisEntity implements Serializable{
    
    private static final long serialVersionUID = 1370692830319429806L;
 
    @Id
    private Long id;
    
//    @Indexed
    private String firstname;
    
//    @Indexed
    private String lastname;
 
    private int age;
 
}
 
cs


Redis 엔티티 설정이다. @RedisHash("person")으로 해당 엔티티가 레디스엔티티임을 명시하였다. 여러글을 읽다가 딱 설명하기 좋은 글이있었다.




RedisEntity라는 엔티티 데이터들을 이후에 무수히 많이 저장이 될것이다. 그래서 이 엔티티들만을 보관하는 하나의 해쉬키 값이 @RedisHash("person")이

되는 것이다. 그리고 이 해쉬 공간에서 각 엔티티들이 person:hash_id 라는 아이디 값을 가지게 된다.(실제로 @Id에 매핑되는 것은 Hash_id) 

이것을 간단히 cli로 보여드리면



사실 key list를 불러오기 위해 " keys * "라는 명령어를 사용할 수 있지만, 이 명령어를 실행시키면 그동안 Redis의 모든 행동은 all stop 됨에 주의하자.


데이터 구조가 이해가 되는가?

이것을 자바의 해쉬 타입으로 지정한다면

HashMap<String,HashMap<String,Person>>의 구조가 되는 것이다. 해쉬의해쉬타입이 된다는 것이 그림에서도 표현이 되있다.



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
package com.spring.redis;
 
import java.util.Arrays;
import java.util.List;
 
import javax.annotation.Resource;
 
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.test.context.junit4.SpringRunner;
 
 
@RunWith(SpringRunner.class)
@SpringBootTest
public class RedisTest {
    
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Autowired
    private RedisRepository repository;
    
//    @Test
//    public void testDataHandling() {
//        
//        redisTemplate.getConnectionFactory().getConnection().info().toString();
//        
//        String key = "yeoseong";
//        String value = "yoon";
//        redisTemplate.opsForValue().set(key, value);
//        String returnValue = (String) redisTemplate.opsForValue().get(key);
//        
//        System.out.println(value);
//    }
    
    @Test
    public void redisRepository() {
        RedisEntity entity = new RedisEntity();
        entity.setFirstname("yeoseong");
        entity.setLastname("yoon");
        entity.setAge(28);
        repository.save(entity);
        RedisEntity findEntity = repository.findByFirstname(entity.getFirstname());
        System.out.println(findEntity.toString());
    }
}
 
cs


저작자표시

'Middleware > Redis' 카테고리의 다른 글

Springboot,Redis - Springboot Redis Nodes Cluster !(레디스 클러스터)  (0) 2019.03.01
Redis - Cluster & Sentinel 차이점 및 Redis에 대해  (0) 2019.02.28
Redis - Sentinel 이란? 설정방법! Redis 고가용성을 위한 방법  (2) 2019.02.27
Redis - Redis 설치 및 설정, 간단한 사용방법  (1) 2019.02.27
posted by 여성게
:

티스토리툴바