...
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
//스프링 클라우드 스트림에 애플리케이션 메시지 브로커로 바인딩 하라고 알린다.
//Source 클래스에 정의된 채널들을 이용해 메시지 브로커와 통신하게 된다.
@EnableBinding(Source.class)
public class Application {
...
브로커에 메시지를 발행해보자
@Component
public class SimpleSourceBean {
private Source source;
//서비스가 사용할 소스 인터페이스 구현을 주입한다.
//특정 메시지 토픽에 대한 모든 통신은 스프링 클라우드 스트림 구조로 발생한다.
@Autowired
public SimpleSourceBean(Source source){
this.source = source;
}
public void publishOrgChange(String action, String orgId){
//POJO 메세지를 발행한다.
OrganizationChangeModel change = new OrganizationChangeModel(
OrganizationChangeModel.class.getTypeName(),
action,
orgId,
UserContext.getCorrelationId());
//Source클래스에서 정의된 채널에서 send() 메서드를 사용해 메세지를 발행한다.
source.output().send(MessageBuilder.withPayload(change).build());
/**
Source 인터페이스는 스프링 클라우드에서 정의한 인터페이스로 output()이라는 단일 메서드를 노출한다.
output 메서드는 MessageChannel 클래스 타입을 반환한다.
send() 메서드는 스프링 MessageBuilder라는 스프링 헬퍼 클래스를 사용해 변환된 Message를 매개변수로 받는다.
*/
}
}
...
stream:
bindings: //srteam.binding은 스클스의 메시지 브로커에 발행하려는 구성의 시작점
output: //output은 채널 이름이며, Source.output()채널에 매핑
destination: orgChangeTopic //메시지 큐(or 토픽) 이름
content-type: application/json //송수신할 메시지 타입 정보(or xml, avro)
kafka: //메세지 버스로 카프카를 사용할 것이라고 스프링에 전달
binder:
zkNodes: localhost //카프카와 주키퍼의 네트워크 위치를 전달
brokers: localhost
실제로 어디서 메시지가 발행되는지 살펴보자
@Service
public class OrganizationService {
@Autowired
private OrganizationRepository orgRepository;
@Autowired
SimpleSourceBean simpleSourceBean;
public Organization getOrg(String organizationId) {
return orgRepository.findById(organizationId).get();
}
public void saveOrg(Organization org){
org.setId( UUID.randomUUID().toString());
orgRepository.save(org);
simpleSourceBean.publishOrgChange("SAVE", org.getId());
//생성, 삭제 등 모든 메서드가 publishOrgChange를 사용한다.
}
...
생산자를 정리해보면
데이터가 변경되면 OrganizationService.saveOrg()
소스(SimpleSourceBean) 에서 메시지가 발행됨
채널은 카프카 토픽에 매핑되는 스프링 클라우드 스트림의 채널 이름이다.
바인더는 카프카 서버에 바인딩 되는 클래스와 구성을 말한다.
이후 카프카 토픽으로 전달되어 처리 진행
소비자
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.annotation.StreamListener;
@EnableBinding(Sink.class) //메시지를 수신가능하도록 Sink 인터페이스에 정의된 채널을 사용
public class Application {
...
@StreamListener(Sink.INPUT) //메시지가 수신될 때마다 실행 된다.
public void loggerSink(OrganizationChangeModel orgChange) {
//매개변수인 OrganizationChangeModel로 자동 역직렬화 된다.
logger.debug("Received an event for organization id {}", orgChange.getOrganizationId());
}
application.yml
stream:
bindings:
input: // input채널을 orgChangeTopic 큐에 매핑한다.
destination: orgChangeTopic
content-type: application/json
group: licensingGroup //소비자 그룹을 정의. 여러 인스턴스가 있을 때 소비자 그룹을 식별해서 하나만 소비하도록 보장(카프카 컨슈머 그룹)
kafka:
binder:
zkNodes: localhost
brokers: localhost
ncp-ganesha의 이벤트 사용은?
import org.springframework.kafka.annotation.EnableKafka;
...
@EnableKafka
public class KafkaConfig {
...
}
Service -> kafkaEventProducer.send
// 원상품 싱크 이벤트 생성.
kafkaEventProducer.send(NcpProdEventType.PRODUCT_SYNC, ProductBaseMsg.builder()
.productIds(Lists.newArrayList(productId))
.build());
KafkaEventProducer
...
public class KafkaEventProducer {
...
@Autowired
private KafkaTemplate<Long, Object> kafkaTemplate;
/**
* 이벤트에 의한 발행.
*
* @param event
* @param <T>
*/
public <T> void send(Event<T> event) {
doSend(event.eventType(), event.key(), event.value(), event.timestamp());
}
...
/**
* {@link Transactional} 메소드 내부에서 호출한 경우 정상적으로 Commit 된 이후에 발행한다.
*
* @param event
* @param <T>
*/
public <T> void sendAfterCommit(Event<T> event) {
doSendAfterCommit(event);
}
...
/**
* 실제 발송 처리.
*
* @param eventType
* @param key
* @param value
* @param <T>
*/
private <T> void doSend(EventType eventType, Long key, T value) {
doSend(eventType, key, value, null);
}
private <T> void doSend(EventType eventType, Long key, T value, Long timestamp) {
...
ProducerRecord<Long, Object> record = new ProducerRecord<>(
eventType.eventName(),
null,
timestamp,
key,
value
);
// async 발행.
kafkaTemplate.send(record);
}
...
// @KafkaListener는 일반적인 경우 사용을 권장하지않는다.
// 정말 사용하고 싶을때는 재처리 정도의 작업일 때 사용하자.
요약
메시지/이벤트 기반 아키텍처
메시징을 사용한 비동기식 통신은 MSA의 중요한 부분이다.
메시징을 사용해서 서비스 확장과 fault-tolerant 하게 만들 수 있다.
스프링 클라우드 스트림
SCS은 간단한 애너테이션을 사용해서 플랫폼을 추상화해 메시지 발행과 소비를 단순화 한다.
SCS의 메세지 소스/싱크(발행/수신)는 애너테이션된 자바 메스드로 브로커에 큐를 발행/수신한다.
8장 목표 : 비동기 메시지를 사용하는 Micro Service를 구현해보자
Event Driven Arch, Message Driven Arch
캐시를 이용한 비동기 예제
데이터베이스 엑세스 비용을 줄이고, 응답 시간을 향상시켜보도록 레디스를 써보자.
동기식 요청
전통적인 동기식 요청-응답 프로그래밍
문제점
A서비스-B서비스가 강결합
쉽게 깨지는 서비스 관계
유연하지 못함
=> 전형적인 거미줄 모양의 의존성 패턴 = 주요 장애지점
메시징을 사용한 변화 전달 방식
4가지 장점이 생긴다.
느슨한 결합
내구성
확장성
유연성
메세지 아키텍처의 단점
Message handling semantics 메시지 처리의 의미론
Message visibility 메세지 가시성
Message choreography 메시지 코레오그래피
스프링 클라우드 스트림
스프링 클라우드 스트림 아키텍처
4가지 컴포넌트
생산자/소비자 코드를 작성해보자
생산자
라이브러리를 추가하자
App이 스프링 클라우드 스트림의 메시지 브로커와 바인딩하도록 지정해주자.
브로커에 메시지를 발행해보자
스프링 클라우드 스트림의 Source와 카프카 브로커/토픽을 매핑해보자 application.yml
실제로 어디서 메시지가 발행되는지 살펴보자
생산자를 정리해보면
데이터가 변경되면
OrganizationService.saveOrg()
소스
(SimpleSourceBean) 에서 메시지가 발행됨채널
은 카프카 토픽에 매핑되는 스프링 클라우드 스트림의 채널 이름이다.바인더
는 카프카 서버에 바인딩 되는 클래스와 구성을 말한다. 이후 카프카 토픽으로 전달되어 처리 진행소비자
application.yml
ncp-ganesha의 이벤트 사용은?
Service -> kafkaEventProducer.send
KafkaEventProducer
요약
메시지/이벤트 기반 아키텍처
스프링 클라우드 스트림