메시지 소비자 구현 (이벤트 서비스)

README.md

@@ -3,7 +3,7 @@
 ## Development Environment
 `Windows 10` `JDK 11.0.6` `SpringBoot 2.3.2.RELEASE` `Maven 3.6.3` `Git 2.22.0.windows.1` `intellij`
 [`Spring Cloud Hoxton.SR6`](https://spring.io/projects/spring-cloud) 에서 `SR8` 로 업그레이드 [`RabbitMQ 3.8.6`](https://www.rabbitmq.com/download.html)
-[`Erlang/OTP 23.0`](https://www.erlang.org/downloads)
+[`Erlang/OTP 23.0`](https://www.erlang.org/downloads) [kafka_2.13-2.6.0](https://kafka.apache.org/downloads)
 
  
 
@@ -206,6 +206,23 @@ HOW TO RUN
 
 ***- Spring Cloud Stream (EDA, 비동기 마이크로서비스 구성)***<br />
 
+```shell script
+HOW TO RUN
+
+--  주키퍼 실행
+C:\kafka_2.13-2.6.0\bin\windows>.\zookeeper-server-start.bat ..\..\config\zookeeper.properties
+
+-- 카프카 실행
+C:\kafka_2.13-2.6.0\bin\windows>.\kafka-server-start.bat ..\..\config\server.properties
+
+-- 카프카 토픽 리스트 조회
+C:\kafka_2.13-2.6.0\bin\windows>.\kafka-topics.bat --list --zookeeper localhost:2181
+__consumer_offsets
+mbChangeTopic
+springCloudBus
+```
+
+
 ---
  
 ***- Sleath, Papertrail, Zipkin (Logging Tracker)***<br />

event-service/pom.xml

@@ -40,10 +40,10 @@
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-config</artifactId>
         </dependency>
-        <dependency>
+        <!--<dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-bus-amqp</artifactId>
-        </dependency>
+        </dependency>-->
         <dependency>
             <groupId>org.springframework.security</groupId>
             <artifactId>spring-security-rsa</artifactId>
@@ -67,6 +67,18 @@
             <version>1.1.1.RELEASE</version>
         </dependency>
 
+        <!-- 스프링 클라우드 스트림 -->
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-stream</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-test</artifactId>

event-service/src/main/java/com/assu/cloud/eventservice/EventServiceApplication.java

@@ -1,12 +1,19 @@
 package com.assu.cloud.eventservice;
 
+import com.assu.cloud.eventservice.event.model.MemberChangeModel;
 import com.assu.cloud.eventservice.utils.CustomContextInterceptor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.security.oauth2.resource.UserInfoRestTemplateFactory;
 import org.springframework.cloud.client.loadbalancer.LoadBalanced;
 import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
 import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Primary;
 import org.springframework.security.oauth2.client.OAuth2RestTemplate;
@@ -19,9 +26,21 @@ import java.util.List;
 @EnableEurekaClient
 @SpringBootApplication
 @EnableFeignClients
-@EnableResourceServer       // 보호 자원으로 설정
+@EnableResourceServer           // 보호 자원으로 설정
+@EnableBinding(Sink.class)      // 이 애플리케이션을 메시지 브로커와 바인딩하도록 스프링 클라우드 스트림 설정
+                                // Sink.class 로 지정 시 해당 서비스가 Sink 클래스에 정의된 채널들을 이용해 메시지 브로커와 통신
 public class EventServiceApplication {
 
+    private static final Logger logger = LoggerFactory.getLogger(EventServiceApplication.class);
+
+    /**
+     * 채널에서 받은 메시지를 MemberChangeModel 이라는 POJO 로 자동 역직렬화
+     * @param mbChange
+     */
+    @StreamListener(Sink.INPUT)     // 메시지가 입력 채널에서 수신될 때마다 이 메서드 실행
+    public void loggerSink(MemberChangeModel mbChange) {
+        logger.info("======= Received an event for organization id {}", mbChange.getUserId());
+    }
     /**
      * 사용자 정의 RestTemplate 빈을 생성하여 토큰 삽입
      * RestTemplate 기반 호출이 수행되기 전 후킹되는 메서드

event-service/src/main/java/com/assu/cloud/eventservice/event/model/MemberChangeModel.java

@@ -0,0 +1,51 @@
+package com.assu.cloud.eventservice.event.model;
+
+/**
+ * 발행될 메시지를 표현하는 POJO
+ */
+public class MemberChangeModel {
+    private String type;
+    private String action;
+    private String userId;
+    private String correlationId;
+
+    public MemberChangeModel(String type, String action, String userId, String correlationId) {
+        // super();
+        this.type = type;
+        this.action = action;
+        this.userId = userId;
+        this.correlationId = correlationId;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
+    public String getAction() {
+        return action;
+    }
+
+    public void setAction(String action) {
+        this.action = action;
+    }
+
+    public String getUserId() {
+        return userId;
+    }
+
+    public void setUserId(String userId) {
+        this.userId = userId;
+    }
+
+    public String getCorrelationId() {
+        return correlationId;
+    }
+
+    public void setCorrelationId(String correlationId) {
+        this.correlationId = correlationId;
+    }
+}

event-service/src/main/resources/application.yaml

@@ -1,2 +1,15 @@
 server:
-  port: 8070
+  port: 8070
+# 스프링 클라우드 스트림 설정
+spring:
+  cloud:
+    stream:
+      bindings:
+        input:   # input 은 채널명, EventServiceApplication 의 Sink.INPUT 채널에 매핑되고, input 채널을 mgChangeTopic 큐에 매핑함
+          destination: mbChangeTopic       # 메시지를 넣은 메시지 큐(토픽) 이름
+          content-type: application/json
+          group: eventGroup   # 메시지를 소비할 소비자 그룹의 이름
+      kafka:    # stream.kafka 는 해당 서비스를 카프카에 바인딩
+        binder:
+          zkNodes: localhost    # zkNodes, brokers 는 스트림에게 카프카와 주키퍼의 네트워크 위치 전달
+          brokers: localhost