Cloudkarafka의 이미지 소스

이 자습서/실험실에서는 다음을 사용하여 매우 간단하고 빠른 실습을 경험하게 됩니다.

자바

스프링부트

아파치 카프카

사육사

요구 사항

Homebrew

Postman으로 테스트 . 자세히 알아보려면 다음을 시청하세요.

분할불

자바 8 설치

$ brew tap adoptopenjdk/openjdk
$ brew cask install adoptopenjdk8

카프카 설치

$ brew install kafka

첫 번째 터미널을 열어 Zookeeper를 시작합니다.

$ zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties

두 번째 터미널을 열어 Kafka를 시작합니다.

$ kafka-server-start /usr/local/etc/kafka/server.properties

세 번째 터미널을 열어 Kafka 주제를 생성합니다.

$ kafka-topics --bootstrap-server localhost:9092 --topic <enter-a-topic> --create --partitions 1 --replication-factor 1

Spring Boot로 Maven 프로젝트 생성

열기 start.spring.io

Apache Kafka 종속성을 위해 Spring 웹 및 Spring 선택

IDE에서 프로젝트 열기(IntelliJ 사용 선호)

Kafka 생산자 및 소비자 구성

이것을 application.properties 폴더의 src/main/resources에 추가하고 하이라이트를 수정합니다.

spring.datasource.url=jdbc:mysql://localhost:3306/<your-db-name>?allowPublicKeyRetrieval=true&useSSL=false&serverTimezone=UTC
spring.datasource.username=<your-db-username>
spring.datasource.password=<your-db-password>

app.topic=<same-topic-created-in-terminal>

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.groupId=<any-group-id>

spring.kafka.consumer.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=<any-group-id>
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.producer.bootstrap-servers=localhost:9092
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer

새 터미널을 열고 주제를 생성합니다.

kafka-topics --bootstrap-server localhost:9092 --topic <same-topic-in-application.properties> --create --partitions 1 --replication-factor 1

Kafka 생산자 및 수신자 서비스 초기화

패키지com.example 내부에 다음 폴더를 만듭니다. $ mkdir controller model repository service

생산자 및 수신자 서비스 생성: $ touch service/KafkaConsumer.java service/KafkaProducer.java

KafkaConsumer에 다음을 추가합니다.

package <enter-your-package-here>;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;
@Service
public class KafkaConsumer {
    public static List<String> messages = new ArrayList<>();
    private final static String topic = "<same-topic-in-resources>";
    private final static String groupId = "<same-group-id>";

    @KafkaListener(topics = topic, groupId = groupId)
    public void listen(String message) {
        messages.add(message);
    }
}

다음 코드를 KafkaProducer에 추가합니다.

package <enter-your-package-here>;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${app.topic}")
    private String topic;

    public void produce(String message) {
        kafkaTemplate.send(topic, message);
    }
}

Kafka API 만들기

Kafka 컨트롤러 파일$ touch controller/kafkaController.java을 만들고 다음을 추가합니다.

package server.controller;

import org.springframework.web.bind.annotation.*;

import server.service.KafkaConsumer;
import server.service.KafkaProducer;

import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;

@RestController
public class KafkaController {
  @Autowired
  private KafkaConsumer consumer;

  @Autowired
  private KafkaProducer producer;

  @PostMapping("/send")
  public void send(@RequestBody String data) {
    producer.produce(data);
  }
  @GetMapping("/receive")
  public List<String> receive() {
    return consumer.messages;
  }

  public KafkaConsumer getConsumer() {
    return consumer;
  }

  public KafkaProducer getProducer() {
    return producer;
  }

  public void setConsumer(KafkaConsumer consumer) {
    this.consumer = consumer;
  }

  public void setProducer(KafkaProducer producer) {
    this.producer = producer;
  }
}

참고: 데이터베이스 구성에 사용하는 모델과 리포지토리도 있지만 이 자습서에서는 필요하지 않습니다.

최종 프로젝트 구조

테스트

생산자로부터 메시지를 보내는 방법에는 Postman과 Terminal의 두 가지가 있습니다.

이전에 주제를 생성하는 데 사용한 세 번째 터미널을 엽니다.

$ kafka-console-consumer --bootstrap-server localhost:9092 --topic <enter-your-topic> --from-beginning

1. 우체부

POST/메서드를 사용하고 JSON 본문과 함께 localhost:8080/send를 입력합니다.

{
   "message": "Hello from Postman"
}

2. 터미널
네 번째 터미널을 열고 다음을 입력합니다.

$ kafka-console-producer --broker-list localhost:9092 --topic <enter-your-topic>
> Hello from the terminal

터미널을 확인하면 우편 배달부와 터미널을 통해 보낸 메시지가 실시간으로 표시됩니다.

축하합니다!

후속 조치 및 Github에서 나와 연결 | |