diff --git a/README.md b/README.md
index ce27ff4..b48e268 100644
--- a/README.md
+++ b/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, 비동기 마이크로서비스 구성)***
+```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)***
diff --git a/event-service/pom.xml b/event-service/pom.xml
index ce128fb..3815d08 100644
--- a/event-service/pom.xml
+++ b/event-service/pom.xml
@@ -40,10 +40,10 @@
org.springframework.cloud
spring-cloud-starter-config
-
+
org.springframework.security
spring-security-rsa
@@ -67,6 +67,18 @@
1.1.1.RELEASE
+
+
+ org.springframework.cloud
+ spring-cloud-stream
+
+
+
+
+ org.springframework.cloud
+ spring-cloud-starter-stream-kafka
+
+
org.springframework.boot
spring-boot-starter-test
diff --git a/event-service/src/main/java/com/assu/cloud/eventservice/EventServiceApplication.java b/event-service/src/main/java/com/assu/cloud/eventservice/EventServiceApplication.java
index 2536a0b..ddb0f5f 100644
--- a/event-service/src/main/java/com/assu/cloud/eventservice/EventServiceApplication.java
+++ b/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 기반 호출이 수행되기 전 후킹되는 메서드
diff --git a/event-service/src/main/java/com/assu/cloud/eventservice/event/model/MemberChangeModel.java b/event-service/src/main/java/com/assu/cloud/eventservice/event/model/MemberChangeModel.java
new file mode 100644
index 0000000..d89d063
--- /dev/null
+++ b/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;
+ }
+}
diff --git a/event-service/src/main/resources/application.yaml b/event-service/src/main/resources/application.yaml
index 3fdae5b..eca6dfc 100644
--- a/event-service/src/main/resources/application.yaml
+++ b/event-service/src/main/resources/application.yaml
@@ -1,2 +1,15 @@
server:
- port: 8070
\ No newline at end of file
+ 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
\ No newline at end of file