Django, SpringBoot-카프카 연결

from kafka import KafkaConsumer

import json
import sys
import six
if sys.version_info >= (3, 12, 0):
    sys.modules["kafka.vendor.six.moves"] = six.moves
import threading

# 카프카 메시지를 읽는 클래스
class MessageConsumer:
    def __init__(self, broker, topic):
        self.broker = broker
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers = self.broker,
            value_deserializer = lambda x: x.decode("utf-8"),
            group_id = "my-group",
            auto_offset_reset = "earliest",
            enable_auto_commit = True
        )
    def receive_message(self):
        try:
            for message in self.consumer:
                result = json.loads(message.value)
                imsi = result["data"]
                doc = {"bid":imsi["bid"], "title":imsi["title"],
                       "author":imsi["author"],
                       "category":imsi["category"],
                       "pages":imsi["pages"],
                       "price":imsi["price"],
                       "published_date":imsi["published_date"],
                       "description":imsi["description"]}
                # MongoDB에 삽입
                conn = MongoClient('127.0.0.1')
                db=conn.cqrs
                collect = db.books
                collect.insert_one(doc)
                conn.close()
        except Exception as exc:
            raise exc

        con.close()

        # 메시지 수신자 생성
        broker = ["localhost:9092"]
        topic = "cqrstopic"
        consumer = MessageConsumer(broker)
        # 스레드로 메서드 호출
        t = threading.Thread(target=consumer.receive_message)
        t.start()

import './App.css';
import {Paper} from "@material-ui/core"
import AddBook from "./AddBook";
import Axios from "axios";

import React, {useEffect, useState} from 'react'

function App() {
 //상태를 생성 - 변수를 생성하고 접근자 함수를 생성
 const [items, setItems] = useState([])
 const [data, setData] = useState([0])

 //화면이 출력되자마자 수행될 함수
 //data가 변경된 경우 동작
 useEffect(() => {
   Axios.get("http://127.0.0.1:8000/cqrs/books/")
   .then((response) => {
     console.log(response.data)
     if(response.data){
       setItems(response.data)
     }else{
       alert("읽기 실패")
     }
   })
 }, [data])

 //데이터 추가를 위한 함수
 const add = (book) => {
     console.log("book : ", book);
     Axios.post("http://127.0.0.1:7000/cqrs/book/", book).then((response) => {
       console.log(response.data)
       if (response.data.bid) {
         alert("저장에 성공했습니다.")
         //데이터가 추가될 때 상태를 변경해서
         //데이터를 다시 출력하도록 합니다.
         setData(1)
       } else {
         alert("코멘트를 저장하지 못했습니다.");
       }
     });
 };

 return (
   <div className="App">
     <Paper style={{ margin: 16 }}>
       <AddBook add = {add}/>
     </Paper>
     {items.map((item, index) => (
      <p key={index}>
        {item.title}
      </p>
    ))}

   </div>
 );
}

export default App;

server:
  port: 7000

spring:
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mariadb://127.0.0.1:3306/springcqrs
    username: root
    password: 0000

jpa:
  hibernate:
    ddl-auto: update
  properties:
    hibernate:
      format_sql: true
      show_sql: true

package com.example.demo;

import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.web.servlet.config.annotation.CorsRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

@Configurable
public class WebConfig  implements WebMvcConfigurer {
    @Override
    public void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/**")
                .allowedOrigins("http:/localhost:3000");
    }
}

package com.gmail.dragonhailstone.dataread;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

@Component
public class KafkaConfiguration {
    @Value("${spring.kafka.bootstrap.servers}")
    private String bootstrapServers;


    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return new DefaultKafkaProducerFactory<>(configs);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}