[Kafka/Redis] Spring for Apache Kafka

1주차 - Kafka 개념
2주차 - Kafka 설치
3주차 - Kafka와 Spring
4주차 - Kafka 메트릭
5주차 - Redis 개념6주차 - Kafka/Redis를 활용한 실시간 트랜잭션 집계 서비스 

 

 

1. 설치 

• Zookeeper 설치 
• tmp 파일에 log 삭제 
• Foreground 모드로 실행

bin/zkServer.sh start-foreground

# kafka 실행
.\bin\windows\kafka-server-start.bat .\config\server.properties
# 토픽 생성
.\bin\windows\kafka-topics.bat --create --topic quickstart-events --bootstrap-server localhost:9092
#토픽 확인
.\bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

• Spring에서 확인

@Bean
public ApplicationRunner runner (KafkaTemplate<String, String> kafkaTemplate) {
    return args -> {
        kafkaTemplate.send("quickstart-event", "Hello Kafka ");
    };
}

@KafkaListener(id = "fastcampus-id", topics = "quickstart-event")
public void listen(String message) {
    System.out.println("Received Message : " + message);
}

 

 

2. 토픽 

• 토픽 명
  - 토픽명은 한번 정하면 바꾸기가 매우 어려움 
• 토픽의 파티션 수
  - 파티션 수를 늘릴 수는 있지만 줄일 수는 없음 
  - 1초당 메시지 발행 수 / Consumer Thread 1개가 1초당 처리하는 메세지 수 
• Retention 시간 

1) 토픽 생성

@Configuration
public class KafkaTopicConfiguration {
	@Bean
	public NewTopic clip2() {
		return TopicBuilder.name("clip2")
				.build();
	}
   
   # 복수로 만들기 
   @Bean
	public KafkaAdmin.NewTopics clip2s() {
		return new KafkaAdmin.NewTopics(
			TopicBuilder.name("clip2-part1")
				.build(),
			TopicBuilder.name("clip2-part2")
				.partitions(3) 👈 수정을 해도 파티션의 정보만 바뀜
				.replicas(1)
				.config(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(1000 * 60 * 60))
				.build()
		);
	}
    
}

 

return args -> {
    Map<String, TopicListing> topics = adminClient.listTopics().namesToListings().get();
    for (String topicName : topics.keySet()) {
        TopicListing topicListing = topics.get(topicName);
        System.out.println(topicListing);

        Map<String, TopicDescription> descriptionMap = adminClient.describeTopics(
            Collections.singletonList(topicName)).all().get();
        System.out.println(descriptionMap);

        if (!topicListing.isInternal()) {
            adminClient.deleteTopics(Collections.singletonList(topicName));
        }
    }

};

토픽 확인 법 

 

 

 

3. Pulblish

1) KafkaTemplate

• 기본적으로 비동기 처리 
• Message 객체를 이용 
• 메세지 헤더로 정보를 제공 (topic, partition_id, message key..)

@Configuration
public class KafkaTemplateConfiguration {
	@Bean
	public KafkaTemplate<String, String> kafkaTemplate() {
		return new KafkaTemplate<>(producerFactory());
	}
	
    # 설정을 커스텀으로 지정하면 해당 configuration으로 실행됨 
	private ProducerFactory<String, String> producerFactory() {
		return new DefaultKafkaProducerFactory<>(producerProps());
	}

	private Map<String, Object> producerProps() {
		Map<String, Object> props = new HashMap<>();

		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		return props;
	}
}

@Bean
public ApplicationRunner runner(KafkaTemplate<String, String> kafkaTemplate) {
	# 템플릿을 이용해서 topic에 메세지 발행함 
    return args -> {
        kafkaTemplate.send("clip3", "Hello, Kafka!");
    };
}

 

사용 <- 커스텀 Producer <- kafkaTemplate

 

2) RoutingKafkaTemplate

전송하는 토픽별로 옵션을 다르게 설정할 수 있음 

✳  Deserializer

@Configuration
public class RoutingKafkaTemplateConfiguration {
	@Bean
	public RoutingKafkaTemplate routingKafkaTemplate() {
		return new RoutingKafkaTemplate(factories());
	}

	private Map<Pattern, ProducerFactory<Object, Object>> factories() {
		Map<Pattern, ProducerFactory<Object, Object>> factories = new HashMap<>();
		factories.put(Pattern.compile("clip3-byte"), byteProducerFactory()); ✅ 
		factories.put(Pattern.compile("clip.*"), defaultProducerFactory());
		return factories;
	}

	private ProducerFactory<Object, Object> byteProducerFactory() {
		Map<String, Object> props = new HashMap<>();
        ✅ 설정을 변경 
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
		return new DefaultKafkaProducerFactory<>(producerProps());
	}

	private ProducerFactory<Object, Object> defaultProducerFactory() {

		return new DefaultKafkaProducerFactory<>(producerProps());
	}

	private Map<String, Object> producerProps() {
		Map<String, Object> props = new HashMap<>();

		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		return props;
	}

}

 

3) ReplyingKafkaTemplate

Consumer가 특정 데이터를 전달 받았는지 여부를 확인할 수 있음 

 

 

4. Consumer

• Message Listener
  Thread-safe하지 않으므로  Listener를 호출하는 스레드에서만 호출
  
  • Record 
  • Batch
• Commit Method - AckMode
  • Record
  • Batch
  • Time
  • Count

• MessageListener Container
  • KafkaMessageListenerContainer
  • ConcurrentMessageListenerContainer
• @KafkaListener
  • SpringBoot에서 기본 세팅됨 
  • KafkaAutoConfiguration
  • Property
  • 메타데이터 (OFFEST, RECEIVED_MESSAGE_KEY, ...) -> ConsumerRecordMetadata
  • PayLoad Validator
• Retrying 
  • Retrying Deliveries
  • RetryStateful

 

Container

// MeassageListener설정
@Configuration
public class MessageListenerContainerConfiguration
{
	@Bean
	public KafkaMessageListenerContainer<String, String> kafkaMessageListenerContainer() {
		ContainerProperties containerProps = new ContainerProperties("clip4");
		containerProps.setGroupId("clip4-container");
		containerProps.setAckMode(ContainerProperties.AckMode.BATCH);
		containerProps.setMessageListener(new DefaultMessageListener());
		return new KafkaMessageListenerContainer<>(containerFactory(), containerProps);
	}

	private ConsumerFactory<String, String> containerFactory() {
		return new DefaultKafkaConsumerFactory<>(props());
	}

	private Map<String, Object> props() {
		Map<String, Object> props = new HashMap<>();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

		return props;
	}

}

 

 [Consumer clientId=consumer-clip4-container-1, groupId=clip4-container] Subscribed to topic(s): clip4

 

 

KafkaListener

@Service
public class ClipConsumer {
	@KafkaListener(id = "clip4-listener-id", topics = "clip4-listener")
	public void listen(String message
		, ConsumerRecordMetadata metadata
		, @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp
		, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic
		, @Header(KafkaHeaders.RECEIVED_PARTITION) int partition) {
		System.out.println("Listener Message = " + message + ", metadata = " + metadata.offset());
	}
}

 

 

Validation

json serialize

@Configuration
public class KafkaJsonTemplateConfiguration {
	@Bean
	public KafkaTemplate<String, Animal> kafkaJsonTemplate() {
		return new KafkaTemplate<>(producerFactory());
	}

	private ProducerFactory<String, Animal> producerFactory() {
		return new DefaultKafkaProducerFactory<>(producerProps());
	}

	private Map<String, Object> producerProps() {
		Map<String, Object> props = new HashMap<>();
		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
		return props;
	}
}

@Configuration
public class KafkaJsonListenerContainerConfiguration implements KafkaListenerConfigurer {
	private final LocalValidatorFactoryBean validator;

	public KafkaJsonListenerContainerConfiguration(LocalValidatorFactoryBean validator) {
		this.validator = validator;
	}

	@Bean
	public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Animal>> kafkaJsonListenerContainerFactory() {
		ConcurrentKafkaListenerContainerFactory<String, Animal> factory = new ConcurrentKafkaListenerContainerFactory<>();
		factory.setConsumerFactory(animalConsumerFactory());
		return factory;
	}

	private ConsumerFactory<String, Animal> animalConsumerFactory() {
		return new DefaultKafkaConsumerFactory<>(
			props(),
			new StringDeserializer(),
			new JsonDeserializer<>(Animal.class)
		);
	}

	private Map<String, Object> props() {
		Map<String, Object> props = new HashMap<>();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonSerializer.class);
		return props;
	}

	@Override
	public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
		registrar.setValidator(validator);
	}
}

public class Animal {
	private final String name;
	private final int age;

	@JsonCreator
	public Animal(@JsonProperty("name") String name,
		@JsonProperty("age") int age) {
		this.name = name;
		this.age = age;
	}

	public String getName() {
		return name;
	}

	public int getAge() {
		return age;
	}
}

 

5. Kafka Stream