Kafka

.gitignore

@@ -18,7 +18,6 @@ bin/
 !**/src/test/**/bin/
 
 ### IntelliJ IDEA ###
-.idea
 *.iws
 *.iml
 *.ipr

ecommerce/src/main/java/com/jy/study/ecommerce/EcommerceApplication.java → a-eureka/src/main/java/com/jy/study/ecommerce/Eureka.java

@@ -6,11 +6,11 @@
 
 @SpringBootApplication
 @EnableEurekaServer  //부트 프로젝트를 유레카 [서버]의 자격으로 등록함
-public class EcommerceApplication {
+public class Eureka {
 
     public static void main(String[] args) {
 
-        SpringApplication.run(EcommerceApplication.class, args);
+        SpringApplication.run(Eureka.class, args);
 
     }
 

build.gradle

@@ -24,6 +24,8 @@ configurations {
 }
 
 
+sourceCompatibility = "11"
+targetCompatibility = "11"
 
 
 //아래 프로젝트들 공통 설정 : subprojects
@@ -33,6 +35,7 @@ subprojects {
     group 'com.jy.study'
     version '1.0-SNAPSHOT'  //빌드시 jar 파일 이름에 들어가게 됨.
     sourceCompatibility = "11"
+    targetCompatibility = "11"
 
     repositories {
         mavenCentral()
@@ -74,3 +77,11 @@ subprojects {
 
 }
 
+
+allprojects {
+    tasks.withType(JavaCompile) {
+        sourceCompatibility = '11'
+        targetCompatibility = '11'
+    }
+}
+

d-order-service/build.gradle

@@ -1,3 +1,4 @@
+
 dependencies {
     developmentOnly 'org.springframework.boot:spring-boot-devtools'
     implementation 'org.springframework.cloud:spring-cloud-starter-netflix-eureka-client'

d-order-service/src/main/java/com/example/dorderservice/controller/OrderController.java

@@ -53,7 +53,7 @@ public ResponseEntity<ResponseOrder> addOrder(@PathVariable String userId
         kafkaOrderProducerService.send("orders" ,orderDto);
 
         /* jpa 이용하는 코드*/
-        orderService.createOrder(orderDto);
+       // orderService.createOrder(orderDto);
 
 
 

d-order-service/src/main/java/com/example/dorderservice/kafka/vo/KafkaOrderDto.java

@@ -3,6 +3,7 @@
 import lombok.Data;
 
 import java.io.Serializable;
+import java.util.Arrays;
 import java.util.List;
 
 @Data
@@ -22,13 +23,13 @@ public KafkaOrderDto(Payload payload) {
 
 
     public void init() {
-        List<Fields> fields = List.of(new Fields("string" , true , "order_id")
+        List<Fields> fields =  List.of(new Fields("string" , true , "order_id")
                 , new Fields("string" , true , "user_id")
                 , new Fields("string" , true , "product_id")
                 , new Fields("int64" , true , "qty")
                 , new Fields("int64" , true , "total_price")
-                , new Fields("int64" , true , "unit_price")
-        );
+                , new Fields("int64" , true , "unit_price"));
+
         this.schema = Schema.builder()
                 .type("struct")
                 .optional(false)

d-order-service/src/main/java/com/example/dorderservice/kafka/vo/Payload.java

@@ -19,7 +19,7 @@ public class Payload {
 
     @Builder
     public Payload(OrderDto orderDto) {
-        this.order_id = orderDto.getOrderId();
+        this.order_id = orderDto.getOrderId().split("-")[0] + "-kafka";
         this.user_id = orderDto.getUserId();
         this.product_id = orderDto.getProductId();
         this.qty = orderDto.getQty();

readme.md

@@ -1,36 +1,35 @@
 # Micro Service 간 통신
 ``` text
-OpenFeign 이용하기
+Kafka 이용하기
 ```
-- userService에서 FeingClient를 이용한다.
-- @FeignClient 인터페이스를 작성한다.
 
-## @FeignClient 인터페이스 작성
-
-```java
-@FeignClient(name="order-service")
-public interface OrderServiceClient {
-
-  @GetMapping("/{userId}/orders")
-  List<ResponseOrder> getOrders(@PathVariable String userId);
-
-}
-```
-
-## 사용
-
-- 빈으로 주입받아 사용한다.
-
-```java
-  private final OrderServiceClient orderServiceClient;
-  //중략
-  List<ResponseOrder> orders = orderServiceClient.getOrders;
-
-```
-
-## 에러처리
-
-- ErrorDecoder interface를 구현한다
-- 이 인터페이스를 Bean으로 등록하면 ExceptionAdvice처럼 동작한다.
-
-- [ErrorDecoder구현 클래스 확인하기](https://github.com/sazzeo/msa-spring-cloud/blob/master/d-user-service/src/main/java/com/example/duserservice/error/FeignErrorDecoder.java)
+## 다른 service 끼리 kafka 통신
+![image](https://user-images.githubusercontent.com/89112466/185573316-9431fb4c-8942-4e5b-9bff-99073c00d291.png)
+- sink 와 source를 service 내부의 kafka class로 구현함
+- kafka connector는 구동되어있어야함
+
+## 동일 레플리카 service 끼리 kafka통신
+![image](https://user-images.githubusercontent.com/89112466/185573329-56414499-0de9-4385-b675-60de4d8cc6f1.png)
+- kafka connector 이용
+- kafka sink로 db에 직접 데이터 insert
+  - kafka sink 정보
+  ```yaml
+  {
+    "name": "my-order-sink-connect",
+    "config": {
+        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
+        "connection.url": "jdbc:postgresql://spring-boot-study.cchzqclxtrke.ap-northeast-2.rds.amazonaws.com:5432/postgres",
+        "connection.user": "db id",
+        "connection.password": "db pw",
+        "auto.create": "true",
+        "auto.evolve": "true",
+        "table.name.format":"msa_study.${topic}",
+        "delete.enabled": "false",
+        "tasks.max": "1",
+        "topics": "orders"
+    }
+  }
+  ```
+
+- 전송 data 포맷
+![image](https://user-images.githubusercontent.com/89112466/185574197-3b59d68c-4a8b-420d-9db8-9991c1a7e436.png)

settings.gradle

@@ -2,7 +2,7 @@
 rootProject.name = 'msa-spring-cloud'
 
 // bootJar
-include 'ecommerce'
+include 'a-eureka'
 include 'user-service'
 
 include 'b-zuul-client-first-service'  //줄 클라이언트 서비스1