Kafka 를 이용한 msa 서비스 통신

Kafka 를 이용한 MSA 서비스간 통신을 진행해 보도록 하겠다.

 

1. 구조

 

2. Order(주문), PG(결제) MSA 프로젝트 생성

 

Order

package com.example.Order;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class OrderController {

	@Autowired
	private OrderService orderService;

	@PostMapping("/order")
	public void order(String key) {
		orderService.order(key);
	}
}

 

package com.example.Order;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

@Service
public class OrderService {

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	public String order(String key) {
		ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send("pay", key);
		return "OK";
	};
}

 

PG

• Order 에서 전송된 Key 가 PG 에 등록된 Key 와 같은지 비교해 하는 서비스

package com.example.PG;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class PGService {

	private final String key = "20230226000000000A";

	@KafkaListener(topics = "pay", groupId = "group-id-pay")
	public void pay(String message) {
		String payKey =  message;

		if (key.equals(payKey)) {
			System.out.println("OK");
		} else {
			System.out.println("PASS");
		}
	}
}

 

3. 주문결제 테스트

 

PostMan 으로 order 호출

 

Order

 

PG

 

 

전송 성공

 

4. PG -> Order 응답 전송

 

이제는 PG 에서 처리 여부에 따른 응답을 Kafka 로 전송해 보자.

 

 

PG

package com.example.PG;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class PGService {

	private final String key = "20230226000000000A";

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	@KafkaListener(topics = "pay", groupId = "group-id-pay")
	public void pay(String message) {
		String payKey =  message;

		if (key.equals(payKey)) {
			kafkaTemplate.send("pay", "OK");
		} else {
			kafkaTemplate.send("pay", "FAIL");
		}
	}
}

 

 

Order

package com.example.Order;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

@Service
public class OrderService {

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	public String order(String key) {
		ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send("pay", key);
		return "OK";
	};

	@KafkaListener(topics = "pay", groupId = "group-id-pay")
	public void pgResult(String message) {
		System.out.println("PG result : " + message);
	}
}

 

어? 그런데 나는 PG 에 결과 값인 "OK" 가 Order 로 응답이 올 줄 알았는데, 실제로 온 은답은 Order 가 보낸 Key 값이 출력 되었다.

 

 

아마도, topics 를 "pay" 로 같이 쓰기 때문에 발생하는 이슈로 보여서 pay-response 로 topic 을 생성 후 다시 시도해 보았다.

 

kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic pay-response

 

Order

 

	@KafkaListener(topics = "pay-response", groupId = "group-id-pay")
	public void pgResult(String message) {
		System.out.println("PG result : " + message);
	}

 

PG

 

	@KafkaListener(topics = "pay", groupId = "group-id-pay")
	public void pay(String message) {
		String payKey =  message;

		if (key.equals(payKey)) {
			kafkaTemplate.send("pay-response", "OK");
		} else {
			kafkaTemplate.send("pay-response", "FAIL");
		}
	}

 

이제는 원하는 결과로 응답이 왔다.

 

 

하나의 Topic 은 하나의 Producer -> Consumer 로 사용 해야 겠다.