실시간으로 주문한 사람에 대한 해당 구에 총 주문 개수를 알고 싶다면?
public class UserDummyDataGenerator {
private static final String DB_URL = "jdbc:mariadb://localhost:3306/user-db";
private static final String DB_USER = "root";
private static final String DB_PASSWORD = "taeyun1215";
private static final String[] ADDRESSES = {"중랑구", "서초구", "강남구"};
private static final String[] ROLES = {"USER", "ADMIN", "SOCIAL"};
public static void main(String[] args) {
try {
Class.forName("org.mariadb.jdbc.Driver");
Connection conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASSWORD);
generateDummyUserData(conn);
conn.close();
} catch (ClassNotFoundException | SQLException e) {
e.printStackTrace();
}
}
private static void generateDummyUserData(Connection conn) throws SQLException {
String insertQuery = "INSERT INTO user (username, password, nickname, phone, email, address, role) VALUES (?, ?, ?, ?, ?, ?, ?)";
Random random = new Random();
// 데이터 생성을 위한 PreparedStatement 준비
try (PreparedStatement pstmt = conn.prepareStatement(insertQuery)) {
// 더미 데이터 개수 (여기서는 10개로 가정)
int numberOfDummyData = 1000;
for (int i = 1; i <= numberOfDummyData; i++) {
pstmt.setString(1, "user" + i); // username
pstmt.setString(2, "pass" + i); // password
pstmt.setString(3, "nickname" + i); // nickname
pstmt.setString(4, "010-1000-100" + i); // phone
pstmt.setString(5, "user" + i + "@example.com"); // email
pstmt.setString(6, ADDRESSES[random.nextInt(ADDRESSES.length)]); // address (랜덤하게 선택)
pstmt.setString(7, ROLES[random.nextInt(ROLES.length)]);
pstmt.executeUpdate();
}
}
}
}
generateDummyUserData
→ 더미 데이터를 생성하는 메소드이다.@WebAdapter
@RestController
@RequiredArgsConstructor
@RequestMapping("/orders")
public class OrderCntSumController {
private final OrderCntSumByAddressUseCase orderCntSumByAddressUseCase;
@PostMapping("/aggregation/get-order-cnt-sum-by-address")
public ResponseEntity<ReturnObject> OrderCntSumByAddress(
@RequestBody OrderCntSumByAddressRequest request
) {
OrderCntSumByAddressCommand command = OrderCntSumByAddressCommand.builder()
.address(request.getAddress())
.build();
long startTime = System.currentTimeMillis();
int orderAmountSum = orderCntSumByAddressUseCase.OrderCntSumByAddress(command);
long endTime = System.currentTimeMillis();
System.out.println("Order-aggregation-service API Call time: " + (endTime - startTime) + " ms");
ReturnObject returnObject = ReturnObject.builder()
.success(true)
.data(orderAmountSum)
.build();
return ResponseEntity.status(HttpStatus.OK).body(returnObject);
}
}
@Slf4j
@UseCase
@Transactional
@RequiredArgsConstructor
public class OrderAmountSumByAddressService implements OrderCntSumByAddressUseCase {
private final GetUserPort getUserPort;
private final GetOrderPort getOrderPort;
@Override
public int OrderCntSumByAddress(OrderCntSumByAddressCommand command) {
List<Long> userIds = getUserPort.getUserIdByAddress(command.getAddress());
List<List<Long>> userPartitionList = null;
if (userIds.size() > 100) {
userPartitionList = partitionList(userIds, 100);
}
List<CompletableFuture<Integer>> futures = new ArrayList<>();
for (List<Long> partitionedList : userPartitionList) {
CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
List<Integer> orderCntList = getOrderPort.getOrderCntByUserIds(partitionedList);
return orderCntList.stream().mapToInt(Integer::intValue).sum();
});
futures.add(future);
}
int orderCntSum = futures.stream()
.map(CompletableFuture::join)
.mapToInt(Integer::intValue)
.sum();
return orderCntSum;
}
private static <T> List<List<T>> partitionList(List<T> list, int partitionSize) {
return IntStream.range(0, list.size())
.boxed()
.collect(Collectors.groupingBy(index -> index / partitionSize))
.values()
.stream()
.map(indices -> indices.stream().map(list::get).collect(Collectors.toList()))
.collect(Collectors.toList());
}
}
getUserPort.getUserIdByAddress(command.getAddress())
, getOrderPort.getOrderCntByUserIds(partitionedList)
이 두 부분은 밑에서 설명하겠다.partitionList(userIds, 100)
여기이다.@ExternalSystemAdapter
@RequiredArgsConstructor
public class UserServiceAdapter implements GetUserPort {
ObjectMapper mapper = new ObjectMapper();
private final CommonHttpClient commonHttpClient;
private final String userServiceUrl = "http://localhost:8000/user-service";
@Override
public List<Long> getUserIdByAddress(String address) {
String url = String.join("/", userServiceUrl, "users/address", address);
try {
String jsonResponse = commonHttpClient.sendGetRequest(url).body();
Map<String, Object> responseMap = mapper.readValue(jsonResponse, new TypeReference<Map<String, Object>>() {});
List<Long> userIds = (List<Long>) responseMap.get("data");
return userIds;
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
@ExternalSystemAdapter
@RequiredArgsConstructor
public class OrderServiceAdapter implements GetOrderPort {
ObjectMapper mapper = new ObjectMapper();
private final CommonHttpClient commonHttpClient;
private final String orderServiceUrl = "http://localhost:8000/order-service";
@Override
public List<Integer> getOrderCntByUserIds(List<Long> userIds) {
String url = String.join("/", orderServiceUrl, "orders/order_count");
try {
Map<String, List<Long>> payload = Collections.singletonMap("userIds", userIds);
String jsonPayload = mapper.writeValueAsString(payload);
Map<String, Object> responseMap = commonHttpClient.sendAsyncPostRequest(url, jsonPayload);
List<Integer> orderCntList = (List<Long>) responseMap.get("data");
return orderCntList;
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
Order-aggregation-service API Call time: 24412 ms
실시간으로 주문한 사람에 대한 해당 구에 총 주문 개수를 알고 싶다면?
를 위 이미지처럼 구현하려면 다음과 같다.dependencies {
// Axon
implementation 'org.axonframework:axon-spring-boot-starter:4.8.0'
}
axon:
axonserver:
servers: localhost:8124 # 비워두면 Axon Server에 연결하지 않도록 설정
enabled: true
serializer:
events: jackson
messages: jackson
general: xstream
@Slf4j
@UseCase
@Transactional
@RequiredArgsConstructor
public class RegisterOrderService implements RegisterOrderUseCase {
private final SaveOrderItemPort saveOrderItemPort;
@Autowired
private CommandGateway commandGateway;
@Override
public void registerOrder(CreateOrderCommand command) {
Order order = Order.builder()
.receiverName(command.getReceiverName())
.receiverPhone(command.getReceiverPhone())
.receiverAddress(command.getReceiverAddress())
.userId(command.getUserId())
.orderStatus(OrderStatus.ORDER_CREATED)
.build();
Order saveOrder = saveOrderPort.saveOrder(order);
// 각종 비즈니스 로직 처리...
command.setOrderId(saveOrder.getOrderId());
commandGateway.sendAndWait(command);
}
}
@Slf4j
@Aggregate
@NoArgsConstructor
public class OrderAggregate {
@AggregateIdentifier
private Long orderId;
private OrderStatus orderStatus; // 주문 상태 (ex: CREATED, COMPLETED, CANCELLED)
private List<OrderCreatedEvent.OrderItemInfo> orderItems;
@CommandHandler
public OrderAggregate(CreateOrderCommand command) {
this.orderId = command.getOrderId();
apply(new OrderCreatedEvent(command.getOrderId(), command.getUserId(), command.getOrderItemInfos()));
}
@EventSourcingHandler
public void on(OrderCreatedEvent event) {
this.orderId = event.getOrderId();
this.orderItems = event.getOrderItems();
this.orderStatus = OrderStatus.ORDER_CREATED;
}
}
@Slf4j
@Component
public class OrderRegisterEventHandler {
@EventHandler
public void handler(
OrderCreatedEvent event,
GetUserPort getUserPort,
InsertOrderCntStatePort insertOrderCntStatePort
) {
String address = getUserPort.getAddressByUserId(event.getUserId());
insertOrderCntStatePort.InsertOrderCntByAddress(address, event.getOrderItems().size());
}
}
getUserPort.getAddressByUserId(event.getUserId())
→ 주문한 사람의 userId값을 매개변수값으로 넣어주면 그 사람의 해당하는 구를 반환해준다.@PersistenceAdapter
@RequiredArgsConstructor
public class OrderCntSumCommandAdapter implements InsertOrderCntStatePort {
private final OrderCntSumByAddressRepo repo;
@Override
public void InsertOrderCntByAddress(String addressName, int increaserderCnt) {
OrderCntSumByAddress orderCntSumByAddress = repo.findByAddress(addressName)
.orElseGet(() -> createNewRecord(addressName));
orderCntSumByAddress.setOrderCnt(orderCntSumByAddress.getOrderCnt() + increaserderCnt);
repo.save(orderCntSumByAddress);
}
private OrderCntSumByAddress createNewRecord(String addressName) {
OrderCntSumByAddress newRecord = OrderCntSumByAddress.builder()
.address(addressName)
.orderCnt(0L)
.build();
return repo.save(newRecord);
}
@WebAdapter
@RestController
@RequiredArgsConstructor
@RequestMapping("/orders")
public class OrderCqrsController {
private final OrderCntSumUseCase orderCntSumUseCase;
@GetMapping("/cqrs/get-order-cnt-sum-by-address/{address}")
public ResponseEntity<ReturnObject> OrderCntSumByAddress(
@PathVariable("address") String address
) {
OrderCntSumByAddressQuery query = OrderCntSumByAddressQuery.builder()
.address(address)
.build();
long startTime = System.currentTimeMillis();
Long orderCntSumByAddress = orderCntSumUseCase.orderCntSumByAddress(query);
long endTime = System.currentTimeMillis();
System.out.println("Order-CQRS-Service API Call time: " + (endTime - startTime) + " ms");
ReturnObject returnObject = ReturnObject.builder()
.success(true)
.data(orderCntSumByAddress)
.build();
return ResponseEntity.status(HttpStatus.OK).body(returnObject);
}
}
OrderCntSumByAddressQuery
→ 클래스 명에 Query를 붙여주면서 해당 클래스가 조회를 하기 위한 클래스라는 것을 의미한다.orderCntSumUseCase
→ UseCase는 Interface라 클래스 명을 포괄적으로 두었고 클래스 안에 메소드를 조금 디테일하게 두었다.@Slf4j
@UseCase
@Transactional
@RequiredArgsConstructor
public class OrderCntSumService implements OrderCntSumUseCase {
private final QueryGateway queryGateway;
@Override
public Long orderCntSumByAddress(OrderCntSumByAddressQuery query) {
return queryGateway.query(query, Long.class).join();
}
}
@PersistenceAdapter
@RequiredArgsConstructor
public class OrderCntSumQueryAdapter {
private final OrderCntSumByAddressRepo repo;
@QueryHandler
public Long orderCntSumByAddress(OrderCntSumByAddressQuery query) {
return repo.findByAddress(query.getAddress())
.map(OrderCntSumByAddress::getOrderCnt)
.orElse(null);
}
}
Order-CQRS-Service API Call time: 926ms