microservice communicate pattern

// Lựa chọn 1: Spring 6 RestClient (Blocking, thiết kế thanh lịch)
@Component
public class UserClient {
   private final RestClient restClient;
 
   public UserClient(@Value("${service.user.url}") String baseUrl) {
       this.restClient = RestClient.builder()
           .baseUrl(baseUrl)
           .defaultHeader("Content-Type", "application/json")
           .build();
   }
 
   public UserDTO getUser(UUID id, String token) {
       return restClient.get()
           .uri("/users/{id}", id)
           .header("Authorization", "Bearer " + token)
           .retrieve()
           // Chỉ 404 mới là "không tìm thấy"; 401/403/400 không nên bị gộp vào đây
           .onStatus(status -> status.value() == 404, (req, res) -> {
               throw new UserNotFoundException(id);
           })
           .body(UserDTO.class);
   }
}
 
// Lựa chọn 2: WebClient (Non-blocking, hướng tới hiệu năng cao)
@Component
public class AsyncUserClient {
   private final WebClient webClient;
 
   public AsyncUserClient(WebClient.Builder builder, @Value("${service.user.url}") String baseUrl) {
       this.webClient = builder.baseUrl(baseUrl).build();
   }
 
   public Mono<UserDTO> getUser(UUID id) {
       return webClient.get()
           .uri("/users/{id}", id)
           .retrieve()
           .bodyToMono(UserDTO.class)
           .timeout(Duration.ofSeconds(3))
           .onErrorResume(WebClientResponseException.NotFound.class,
               e -> Mono.empty());
   }
}

// High-performance binary protocol
// Proto definition → generated stubs
@GrpcService
public class UserGrpcService extends UserServiceGrpc.UserServiceImplBase {
   @Override
   public void getUser(GetUserRequest request, StreamObserver<UserResponse> observer) {
       User user = userService.findById(request.getId());
       observer.onNext(toProto(user));
       observer.onCompleted();
   }
}
 
// Client
@Component
public class UserGrpcClient {
   private final UserServiceGrpc.UserServiceBlockingStub stub;
 
   public UserResponse getUser(String id) {
       return stub.withDeadlineAfter(3, TimeUnit.SECONDS)
           .getUser(GetUserRequest.newBuilder().setId(id).build());
   }
}

@Service
public class OrderService {
   public Order createOrder(CreateOrderDTO dto) {
       // 1. Lưu trạng thái vào core domain
       Order order = repo.save(map(dto));
 
       // 2. Bắn sự kiện ra thế giới bên ngoài (Infrastructure layer)
       // OrderService không cần biết ai sẽ xử lý (Inventory, Notification, etc.)
       rabbitTemplate.convertAndSend(
           "order.exchange",
           "order.created",
           new OrderCreatedEvent(order.getId())
       );
 
       return order;
   }
}

public PaymentResult requestPayment(PaymentRequest request) {
   Message reply = rabbitTemplate.sendAndReceive(
       "payment.exchange",
       "payment.process",
       MessageBuilder.withBody(serialize(request))
           .setCorrelationId(UUID.randomUUID().toString())
           .build()
   );
   if (reply == null) { // hết replyTimeout mà chưa có phản hồi
       throw new PaymentTimeoutException(request);
   }
   return deserialize(reply.getBody(), PaymentResult.class);
}

// Dependency
// implementation 'io.github.resilience4j:resilience4j-spring-boot3'
 
@Service
public class PaymentClient {
 
   @CircuitBreaker(name = "paymentService", fallbackMethod = "paymentFallback")
   public PaymentResult processPayment(PaymentRequest request) {
       return restClient.post()
           .uri("/payments")
           .body(request)
           .retrieve()
           .body(PaymentResult.class);
   }
 
   // Lưới an toàn: hứng lỗi khi cầu dao mở hoặc mạng đứt
   private PaymentResult paymentFallback(PaymentRequest request, Throwable t) {
       log.warn("Payment service gián đoạn. Chuyển vào hàng đợi: {}", t.getMessage());
       rabbitTemplate.convertAndSend("payment.retry.queue", request);
       return PaymentResult.pending("Đang chờ xử lý");
   }
}

resilience4j:
  circuitbreaker:
    instances:
      paymentService:
        sliding-window-size: 10
        failure-rate-threshold: 50 # open after 50% failures
        wait-duration-in-open-state: 30s # wait before half-open
        permitted-number-of-calls-in-half-open-state: 3
        slow-call-duration-threshold: 2s
        slow-call-rate-threshold: 80

@Retry(name = "externalApi", fallbackMethod = "retryFallback")
public DataDTO fetchData(String id) {
   return restClient.get().uri("/data/{id}", id).retrieve().body(DataDTO.class);
}

resilience4j:
  retry:
    instances:
      externalApi:
        max-attempts: 3
        wait-duration: 1s
        enable-exponential-backoff: true # thiếu dòng này thì multiplier bị bỏ qua
        exponential-backoff-multiplier: 2
        retry-exceptions:
          # RestClient bọc IOException trong ResourceAccessException,
          # mà Resilience4j chỉ so kiểu của chính exception, không xét cause
          - org.springframework.web.client.ResourceAccessException
          - org.springframework.web.client.HttpServerErrorException
        ignore-exceptions:
          - com.example.BusinessException

@RateLimiter(name = "thirdPartyApi")
public Response callThirdParty(Request request) {
   return externalClient.call(request);
}

resilience4j:
  ratelimiter:
    instances:
      thirdPartyApi:
        limit-for-period: 100
        limit-refresh-period: 1s
        timeout-duration: 500ms

// THREADPOOL bulkhead chỉ áp được cho method trả về CompletionStage/CompletableFuture
@Bulkhead(name = "heavyOperation", type = Bulkhead.Type.THREADPOOL)
public CompletableFuture<Result> processHeavy(Request request) {
   return CompletableFuture.completedFuture(heavyService.process(request));
}

resilience4j:
  bulkhead: # cấu hình cho type SEMAPHORE (mặc định)
    instances:
      heavyOperation:
        max-concurrent-calls: 10
        max-wait-duration: 500ms
  thread-pool-bulkhead: # cấu hình cho type THREADPOOL như ví dụ trên
    instances:
      heavyOperation:
        max-thread-pool-size: 5
        core-thread-pool-size: 3
        queue-capacity: 20

// Thứ tự annotation trong code không quan trọng; mặc định Retry bọc ngoài CircuitBreaker, rồi tới RateLimiter
@CircuitBreaker(name = "backend")
@Retry(name = "backend")
@RateLimiter(name = "backend")
public Response callBackend(Request request) {
   return client.call(request);
}

# Service A (gateway) — total budget: 5s
service-b:
  timeout: 3s # leaves 2s for own processing
 
# Service B — budget from A: 3s
service-c:
  timeout: 1.5s # leaves 1.5s for own processing

@GetMapping("/aggregate")
public AggregateDTO aggregate(@RequestHeader("X-Request-Deadline") Optional<Instant> deadline) {
   // Nếu không có, gán ngân sách tổng là 5 giây
   Instant myDeadline = deadline.orElse(Instant.now().plusSeconds(5));
   Duration remaining = Duration.between(Instant.now(), myDeadline);
 
   if (remaining.isNegative() || remaining.isZero()) {
       // TimeoutException của JDK là checked exception, không throw thẳng được ở đây
       throw new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT, "Ngân sách thời gian đã cạn, huỷ xử lý.");
   }
 
   // Chỉ cấp cho các lời gọi hạ nguồn phần thời gian còn lại,
   // và truyền tiếp deadline mới qua header X-Request-Deadline cho service sau
   Duration downstreamBudget = remaining.dividedBy(2);
   return client.getData(downstreamBudget);
}

# Service registers itself
eureka:
  client:
    service-url:
      defaultZone: http://localhost:8087/eureka/
  instance:
    prefer-ip-address: true
    instance-id: ${spring.application.name}:${server.port}

// Load-balanced RestTemplate via service name (RestClient.Builder cũng gắn @LoadBalanced được)
@Bean
@LoadBalanced
public RestTemplate restTemplate() {
   return new RestTemplate();
}
 
// Call by service name (Eureka resolves to IP:port)
restTemplate.getForObject("http://user-service/users/{id}", UserDTO.class, id);

// When Eureka is unavailable — query MongoDB for registered services
public String resolveServiceUrl(String serviceName) {
   try {
       return eurekaClient.getNextServerFromEureka(serviceName, false).getHomePageUrl();
   } catch (Exception e) {
       // Fallback to MongoDB registry
       ServiceRegistration reg = mongoTemplate.findOne(
           Query.query(Criteria.where("serviceName").is(serviceName).and("status").is("UP")),
           ServiceRegistration.class);
       return reg != null ? reg.getUrl() : null;
   }
}

management:
  tracing:
    sampling:
      probability: 1.0 # 100% in dev, lower in prod (Spring Boot defaults to 0.1)
  otlp:
    tracing:
      endpoint: http://localhost:4318/v1/traces

// Auto-propagated via Micrometer Tracing
// HTTP headers: traceparent, tracestate (W3C format)
// RabbitMQ: message headers, nhưng phải bật observation cho RabbitTemplate/listener
// (spring.rabbitmq.template.observation-enabled=true, spring.rabbitmq.listener.simple.observation-enabled=true)
 
// Manual span creation
@Component
@RequiredArgsConstructor
public class CustomTracing {
   private final Tracer tracer;
 
   public void tracedOperation() {
       Span span = tracer.nextSpan().name("custom-operation").start();
       try (Tracer.SpanInScope ws = tracer.withSpan(span)) {
           doWork();
           span.tag("result", "success");
       } catch (Exception e) {
           span.error(e);
           throw e;
       } finally {
           span.end();
       }
   }
}

# Spring Cloud Gateway (or custom proxy)
spring:
  cloud:
    gateway:
      routes:
        - id: user-service
          uri: lb://user-service
          predicates:
            - Path=/api/users/**
          filters:
            - StripPrefix=1
            - AddRequestHeader=X-Internal, true

@GetMapping("/dashboard")
public DashboardDTO getDashboard(@AuthenticationPrincipal JwtPayload user) {
   CompletableFuture<UserProfile> profile = CompletableFuture.supplyAsync(
       () -> userClient.getProfile(user.getSub()), executor);
 
   CompletableFuture<List<Order>> orders = CompletableFuture.supplyAsync(
       () -> orderClient.getRecent(user.getSub(), 5), executor);
 
   CompletableFuture<Notifications> notifications = CompletableFuture.supplyAsync(
       () -> notifClient.getUnread(user.getSub()), executor);
 
   // Chờ tất cả cùng hoàn thành
   CompletableFuture.allOf(profile, orders, notifications).join();
 
   return new DashboardDTO(profile.join(), orders.join(), notifications.join());
}