Java学习第21天 - 微服务架构设计

学习目标

• 掌握微服务架构的核心概念和设计原则
• 学习服务拆分策略和边界划分
• 掌握微服务间的通信机制
• 了解服务注册与发现机制
• 学习API网关的设计与实现
• 掌握微服务的数据管理策略

一、微服务架构基础

1. 微服务架构概念

// 1. 微服务架构示例 - 用户服务
@RestController
@RequestMapping("/api/users")
@Validated
public class UserController {
    
    @Autowired
    private UserService userService;
    
    @Autowired
    private OrderServiceClient orderServiceClient;
    
    @GetMapping("/{id}")
    public ResponseEntity getUserById(@PathVariable Long id) {
        UserDTO user = userService.getUserById(id);
        return ResponseEntity.ok(user);
    }
    
    @GetMapping("/{id}/orders")
    public ResponseEntity> getUserOrders(@PathVariable Long id) {
        List orders = orderServiceClient.getOrdersByUserId(id);
        return ResponseEntity.ok(orders);
    }
    
    @PostMapping
    public ResponseEntity createUser(@Valid @RequestBody CreateUserRequest request) {
        UserDTO user = userService.createUser(request);
        return ResponseEntity.status(HttpStatus.CREATED).body(user);
    }
}

// 2. 用户服务实现
@Service
@Transactional
public class UserService {
    
    @Autowired
    private UserRepository userRepository;
    
    @Autowired
    private NotificationServiceClient notificationServiceClient;
    
    public UserDTO getUserById(Long id) {
        User user = userRepository.findById(id)
            .orElseThrow(() -> new UserNotFoundException("User not found with id: " + id));
        
        return UserDTO.builder()
            .id(user.getId())
            .username(user.getUsername())
            .email(user.getEmail())
            .status(user.getStatus())
            .createdAt(user.getCreatedAt())
            .build();
    }
    
    public UserDTO createUser(CreateUserRequest request) {
        // 检查用户名是否已存在
        if (userRepository.existsByUsername(request.getUsername())) {
            throw new UsernameAlreadyExistsException("Username already exists");
        }
        
        // 创建用户
        User user = User.builder()
            .username(request.getUsername())
            .email(request.getEmail())
            .password(passwordEncoder.encode(request.getPassword()))
            .status(UserStatus.ACTIVE)
            .createdAt(LocalDateTime.now())
            .build();
        
        user = userRepository.save(user);
        
        // 发送欢迎通知
        notificationServiceClient.sendWelcomeNotification(user.getEmail());
        
        return convertToDTO(user);
    }
    
    private UserDTO convertToDTO(User user) {
        return UserDTO.builder()
            .id(user.getId())
            .username(user.getUsername())
            .email(user.getEmail())
            .status(user.getStatus())
            .createdAt(user.getCreatedAt())
            .build();
    }
}

2. 服务拆分策略

// 1. 领域驱动设计 - 订单服务
@Entity
@Table(name = "orders")
public class Order {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(name = "user_id", nullable = false)
    private Long userId;
    
    @Column(name = "order_number", unique = true, nullable = false)
    private String orderNumber;
    
    @Enumerated(EnumType.STRING)
    @Column(name = "status", nullable = false)
    private OrderStatus status;
    
    @Column(name = "total_amount", precision = 10, scale = 2)
    private BigDecimal totalAmount;
    
    @OneToMany(mappedBy = "order", cascade = CascadeType.ALL, fetch = FetchType.LAZY)
    private List items;
    
    @Column(name = "created_at")
    private LocalDateTime createdAt;
    
    @Column(name = "updated_at")
    private LocalDateTime updatedAt;
    
    // 业务方法
    public void addItem(OrderItem item) {
        if (this.items == null) {
            this.items = new ArrayList<>();
        }
        item.setOrder(this);
        this.items.add(item);
        calculateTotalAmount();
    }
    
    public void calculateTotalAmount() {
        this.totalAmount = items.stream()
            .map(OrderItem::getSubtotal)
            .reduce(BigDecimal.ZERO, BigDecimal::add);
    }
    
    public void confirm() {
        if (this.status != OrderStatus.PENDING) {
            throw new IllegalStateException("Only pending orders can be confirmed");
        }
        this.status = OrderStatus.CONFIRMED;
        this.updatedAt = LocalDateTime.now();
    }
    
    public void cancel() {
        if (this.status == OrderStatus.SHIPPED || this.status == OrderStatus.DELIVERED) {
            throw new IllegalStateException("Cannot cancel shipped or delivered orders");
        }
        this.status = OrderStatus.CANCELLED;
        this.updatedAt = LocalDateTime.now();
    }
}

// 2. 订单服务接口
@Service
@Transactional
public class OrderService {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private InventoryServiceClient inventoryServiceClient;
    
    @Autowired
    private PaymentServiceClient paymentServiceClient;
    
    public OrderDTO createOrder(CreateOrderRequest request) {
        // 验证库存
        for (CreateOrderItemRequest itemRequest : request.getItems()) {
            boolean inStock = inventoryServiceClient.checkStock(itemRequest.getProductId(), itemRequest.getQuantity());
            if (!inStock) {
                throw new InsufficientStockException("Product " + itemRequest.getProductId() + " is out of stock");
            }
        }
        
        // 创建订单
        Order order = Order.builder()
            .userId(request.getUserId())
            .orderNumber(generateOrderNumber())
            .status(OrderStatus.PENDING)
            .createdAt(LocalDateTime.now())
            .build();
        
        // 添加订单项
        for (CreateOrderItemRequest itemRequest : request.getItems()) {
            OrderItem item = OrderItem.builder()
                .productId(itemRequest.getProductId())
                .quantity(itemRequest.getQuantity())
                .unitPrice(itemRequest.getUnitPrice())
                .build();
            order.addItem(item);
        }
        
        order = orderRepository.save(order);
        
        // 预留库存
        inventoryServiceClient.reserveStock(order.getId(), request.getItems());
        
        return convertToDTO(order);
    }
    
    public OrderDTO confirmOrder(Long orderId) {
        Order order = orderRepository.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException("Order not found"));
        
        order.confirm();
        order = orderRepository.save(order);
        
        // 处理支付
        PaymentRequest paymentRequest = PaymentRequest.builder()
            .orderId(orderId)
            .amount(order.getTotalAmount())
            .build();
        
        PaymentResponse paymentResponse = paymentServiceClient.processPayment(paymentRequest);
        
        if (paymentResponse.isSuccess()) {
            order.setStatus(OrderStatus.PAID);
            orderRepository.save(order);
        }
        
        return convertToDTO(order);
    }
    
    private String generateOrderNumber() {
        return "ORD" + System.currentTimeMillis() + RandomUtils.nextInt(1000, 9999);
    }
    
    private OrderDTO convertToDTO(Order order) {
        return OrderDTO.builder()
            .id(order.getId())
            .userId(order.getUserId())
            .orderNumber(order.getOrderNumber())
            .status(order.getStatus())
            .totalAmount(order.getTotalAmount())
            .createdAt(order.getCreatedAt())
            .build();
    }
}

二、服务注册与发现

1. Eureka服务注册中心

// 1. Eureka服务端配置
@SpringBootApplication
@EnableEurekaServer
public class EurekaServerApplication {
    public static void main(String[] args) {
        SpringApplication.run(EurekaServerApplication.class, args);
    }
}

// 2. Eureka客户端配置
@SpringBootApplication
@EnableEurekaClient
@EnableDiscoveryClient
public class UserServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(UserServiceApplication.class, args);
    }
}

// 3. 服务发现客户端
@Service
public class ServiceDiscoveryClient {
    
    @Autowired
    private DiscoveryClient discoveryClient;
    
    @Autowired
    private RestTemplate restTemplate;
    
    public String getServiceUrl(String serviceName) {
        List instances = discoveryClient.getInstances(serviceName);
        if (instances.isEmpty()) {
            throw new ServiceNotFoundException("No instances found for service: " + serviceName);
        }
        
        ServiceInstance instance = instances.get(0);
        return instance.getUri().toString();
    }
    
    public  T callService(String serviceName, String path, Class responseType) {
        String serviceUrl = getServiceUrl(serviceName);
        String url = serviceUrl + path;
        
        return restTemplate.getForObject(url, responseType);
    }
    
    @Bean
    @LoadBalanced
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }
}

2. Consul服务发现

// 1. Consul配置
@Configuration
@EnableDiscoveryClient
public class ConsulConfig {
    
    @Value("${spring.cloud.consul.host:localhost}")
    private String consulHost;
    
    @Value("${spring.cloud.consul.port:8500}")
    private int consulPort;
    
    @Bean
    public ConsulClient consulClient() {
        return new ConsulClient(consulHost, consulPort);
    }
    
    @Bean
    public ConsulServiceRegistry consulServiceRegistry(ConsulClient consulClient) {
        return new ConsulServiceRegistry(consulClient, new ConsulRegistration());
    }
}

// 2. 服务健康检查
@Component
public class ServiceHealthChecker {
    
    @Autowired
    private ConsulClient consulClient;
    
    @Scheduled(fixedRate = 30000) // 每30秒检查一次
    public void checkServiceHealth() {
        try {
            // 检查数据库连接
            boolean dbHealthy = checkDatabaseHealth();
            
            // 检查Redis连接
            boolean redisHealthy = checkRedisHealth();
            
            // 更新服务健康状态
            updateServiceHealth(dbHealthy && redisHealthy);
            
        } catch (Exception e) {
            log.error("Health check failed", e);
            updateServiceHealth(false);
        }
    }
    
    private boolean checkDatabaseHealth() {
        // 数据库健康检查逻辑
        return true;
    }
    
    private boolean checkRedisHealth() {
        // Redis健康检查逻辑
        return true;
    }
    
    private void updateServiceHealth(boolean healthy) {
        // 更新Consul中的服务健康状态
        String serviceId = "user-service";
        String checkId = "user-service-health";
        
        if (healthy) {
            consulClient.agentCheckPass(checkId);
        } else {
            consulClient.agentCheckFail(checkId);
        }
    }
}

三、服务间通信

1. RESTful API通信

// 1. Feign客户端
@FeignClient(name = "order-service", path = "/api/orders")
public interface OrderServiceClient {
    
    @GetMapping("/user/{userId}")
    List getOrdersByUserId(@PathVariable("userId") Long userId);
    
    @PostMapping
    OrderDTO createOrder(@RequestBody CreateOrderRequest request);
    
    @GetMapping("/{orderId}")
    OrderDTO getOrderById(@PathVariable("orderId") Long orderId);
    
    @PutMapping("/{orderId}/status")
    void updateOrderStatus(@PathVariable("orderId") Long orderId, 
                          @RequestBody UpdateOrderStatusRequest request);
}

// 2. 服务间调用服务
@Service
public class OrderIntegrationService {
    
    @Autowired
    private OrderServiceClient orderServiceClient;
    
    @Autowired
    private CircuitBreakerFactory circuitBreakerFactory;
    
    public List getUserOrders(Long userId) {
        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("order-service");
        
        return circuitBreaker.run(() -> {
            try {
                return orderServiceClient.getOrdersByUserId(userId);
            } catch (Exception e) {
                log.error("Failed to get orders for user: {}", userId, e);
                throw new ServiceCallException("Failed to get orders", e);
            }
        }, throwable -> {
            log.warn("Order service is unavailable, returning empty list");
            return Collections.emptyList();
        });
    }
    
    public OrderDTO createOrder(CreateOrderRequest request) {
        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("order-service");
        
        return circuitBreaker.run(() -> {
            try {
                return orderServiceClient.createOrder(request);
            } catch (Exception e) {
                log.error("Failed to create order for user: {}", request.getUserId(), e);
                throw new ServiceCallException("Failed to create order", e);
            }
        }, throwable -> {
            log.error("Order service is unavailable, cannot create order");
            throw new ServiceUnavailableException("Order service is currently unavailable");
        });
    }
}

2. 消息队列通信

// 1. RabbitMQ消息发送
@Service
public class MessagePublisher {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Autowired
    private ObjectMapper objectMapper;
    
    public void publishUserCreatedEvent(UserCreatedEvent event) {
        try {
            String message = objectMapper.writeValueAsString(event);
            rabbitTemplate.convertAndSend("user.exchange", "user.created", message);
            log.info("Published user created event: {}", event.getUserId());
        } catch (Exception e) {
            log.error("Failed to publish user created event", e);
            throw new MessagePublishException("Failed to publish user created event", e);
        }
    }
    
    public void publishOrderCreatedEvent(OrderCreatedEvent event) {
        try {
            String message = objectMapper.writeValueAsString(event);
            rabbitTemplate.convertAndSend("order.exchange", "order.created", message);
            log.info("Published order created event: {}", event.getOrderId());
        } catch (Exception e) {
            log.error("Failed to publish order created event", e);
            throw new MessagePublishException("Failed to publish order created event", e);
        }
    }
}

// 2. 消息消费者
@Component
@RabbitListener(queues = "user.created.queue")
public class UserCreatedEventListener {
    
    @Autowired
    private NotificationService notificationService;
    
    @Autowired
    private AuditService auditService;
    
    @RabbitHandler
    public void handleUserCreated(UserCreatedEvent event) {
        try {
            log.info("Received user created event: {}", event.getUserId());
            
            // 发送欢迎邮件
            notificationService.sendWelcomeEmail(event.getEmail());
            
            // 记录审计日志
            auditService.logUserCreation(event.getUserId(), event.getUsername());
            
        } catch (Exception e) {
            log.error("Failed to process user created event", e);
            // 可以发送到死信队列或重试队列
            throw new RuntimeException("Failed to process user created event", e);
        }
    }
}

// 3. 事件定义
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class UserCreatedEvent {
    private Long userId;
    private String username;
    private String email;
    private LocalDateTime createdAt;
}

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class OrderCreatedEvent {
    private Long orderId;
    private Long userId;
    private String orderNumber;
    private BigDecimal totalAmount;
    private LocalDateTime createdAt;
}

四、API网关设计

1. Spring Cloud Gateway配置

// 1. 网关配置
@Configuration
@EnableWebFlux
public class GatewayConfig {
    
    @Bean
    public RouteLocator customRouteLocator(RouteLocatorBuilder builder) {
        return builder.routes()
            // 用户服务路由
            .route("user-service", r -> r.path("/api/users/**")
                .filters(f -> f
                    .addRequestHeader("X-Gateway", "Spring-Cloud-Gateway")
                    .addResponseHeader("X-Response-Time", String.valueOf(System.currentTimeMillis()))
                    .circuitBreaker(config -> config
                        .setName("user-service-circuit-breaker")
                        .setFallbackUri("forward:/fallback/user-service")))
                .uri("lb://user-service"))
            
            // 订单服务路由
            .route("order-service", r -> r.path("/api/orders/**")
                .filters(f -> f
                    .addRequestHeader("X-Gateway", "Spring-Cloud-Gateway")
                    .addResponseHeader("X-Response-Time", String.valueOf(System.currentTimeMillis()))
                    .circuitBreaker(config -> config
                        .setName("order-service-circuit-breaker")
                        .setFallbackUri("forward:/fallback/order-service")))
                .uri("lb://order-service"))
            
            // 产品服务路由
            .route("product-service", r -> r.path("/api/products/**")
                .filters(f -> f
                    .addRequestHeader("X-Gateway", "Spring-Cloud-Gateway")
                    .addResponseHeader("X-Response-Time", String.valueOf(System.currentTimeMillis()))
                    .circuitBreaker(config -> config
                        .setName("product-service-circuit-breaker")
                        .setFallbackUri("forward:/fallback/product-service")))
                .uri("lb://product-service"))
            
            .build();
    }
    
    @Bean
    public GlobalFilter customGlobalFilter() {
        return (exchange, chain) -> {
            ServerHttpRequest request = exchange.getRequest();
            String path = request.getURI().getPath();
            
            // 记录请求日志
            log.info("Gateway request: {} {}", request.getMethod(), path);
            
            // 添加请求ID
            String requestId = UUID.randomUUID().toString();
            ServerHttpRequest mutatedRequest = request.mutate()
                .header("X-Request-ID", requestId)
                .build();
            
            return chain.filter(exchange.mutate().request(mutatedRequest).build());
        };
    }
}

// 2. 自定义过滤器
@Component
public class AuthenticationFilter implements GlobalFilter, Ordered {
    
    @Autowired
    private JwtTokenProvider jwtTokenProvider;
    
    @Override
    public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        String path = request.getURI().getPath();
        
        // 跳过认证的路径
        if (isPublicPath(path)) {
            return chain.filter(exchange);
        }
        
        // 获取JWT token
        String token = getTokenFromRequest(request);
        if (token == null) {
            return unauthorized(exchange);
        }
        
        // 验证token
        if (!jwtTokenProvider.validateToken(token)) {
            return unauthorized(exchange);
        }
        
        // 添加用户信息到请求头
        String username = jwtTokenProvider.getUsernameFromToken(token);
        ServerHttpRequest mutatedRequest = request.mutate()
            .header("X-User-Name", username)
            .build();
        
        return chain.filter(exchange.mutate().request(mutatedRequest).build());
    }
    
    private boolean isPublicPath(String path) {
        return path.startsWith("/api/auth/") || 
               path.startsWith("/api/public/") ||
               path.equals("/health");
    }
    
    private String getTokenFromRequest(ServerHttpRequest request) {
        String bearerToken = request.getHeaders().getFirst("Authorization");
        if (bearerToken != null && bearerToken.startsWith("Bearer ")) {
            return bearerToken.substring(7);
        }
        return null;
    }
    
    private Mono unauthorized(ServerWebExchange exchange) {
        ServerHttpResponse response = exchange.getResponse();
        response.setStatusCode(HttpStatus.UNAUTHORIZED);
        response.getHeaders().add("Content-Type", "application/json");
        
        String body = "{"error":"Unauthorized","message":"Invalid or missing token"}";
        DataBuffer buffer = response.bufferFactory().wrap(body.getBytes());
        return response.writeWith(Mono.just(buffer));
    }
    
    @Override
    public int getOrder() {
        return -1; // 高优先级
    }
}

// 3. 限流过滤器
@Component
public class RateLimitFilter implements GlobalFilter, Ordered {
    
    private final RedisTemplate redisTemplate;
    private final RateLimiter rateLimiter;
    
    public RateLimitFilter(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.rateLimiter = RateLimiter.create(100.0); // 每秒100个请求
    }
    
    @Override
    public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        String clientIp = getClientIp(request);
        
        // 基于IP的限流
        String key = "rate_limit:" + clientIp;
        String count = redisTemplate.opsForValue().get(key);
        
        if (count == null) {
            redisTemplate.opsForValue().set(key, "1", Duration.ofMinutes(1));
        } else {
            int currentCount = Integer.parseInt(count);
            if (currentCount >= 100) { // 每分钟100个请求
                return tooManyRequests(exchange);
            }
            redisTemplate.opsForValue().increment(key);
        }
        
        return chain.filter(exchange);
    }
    
    private String getClientIp(ServerHttpRequest request) {
        String xForwardedFor = request.getHeaders().getFirst("X-Forwarded-For");
        if (xForwardedFor != null && !xForwardedFor.isEmpty()) {
            return xForwardedFor.split(",")[0].trim();
        }
        return request.getRemoteAddress().getAddress().getHostAddress();
    }
    
    private Mono tooManyRequests(ServerWebExchange exchange) {
        ServerHttpResponse response = exchange.getResponse();
        response.setStatusCode(HttpStatus.TOO_MANY_REQUESTS);
        response.getHeaders().add("Content-Type", "application/json");
        
        String body = "{"error":"Too Many Requests","message":"Rate limit exceeded"}";
        DataBuffer buffer = response.bufferFactory().wrap(body.getBytes());
        return response.writeWith(Mono.just(buffer));
    }
    
    @Override
    public int getOrder() {
        return -2; // 比认证过滤器优先级更高
    }
}

五、微服务数据管理

1. 数据库分离策略

// 1. 用户服务数据访问层
@Repository
public class UserRepository extends JpaRepository {
    
    @Query("SELECT u FROM User u WHERE u.username = :username")
    Optional findByUsername(@Param("username") String username);
    
    @Query("SELECT u FROM User u WHERE u.email = :email")
    Optional findByEmail(@Param("email") String email);
    
    @Query("SELECT u FROM User u WHERE u.status = :status")
    List findByStatus(@Param("status") UserStatus status);
    
    @Modifying
    @Query("UPDATE User u SET u.lastLoginAt = :lastLoginAt WHERE u.id = :id")
    void updateLastLoginAt(@Param("id") Long id, @Param("lastLoginAt") LocalDateTime lastLoginAt);
}

// 2. 订单服务数据访问层
@Repository
public class OrderRepository extends JpaRepository {
    
    @Query("SELECT o FROM Order o WHERE o.userId = :userId ORDER BY o.createdAt DESC")
    List findByUserIdOrderByCreatedAtDesc(@Param("userId") Long userId);
    
    @Query("SELECT o FROM Order o WHERE o.status = :status AND o.createdAt >= :startDate")
    List findByStatusAndCreatedAtAfter(@Param("status") OrderStatus status, 
                                            @Param("startDate") LocalDateTime startDate);
    
    @Query("SELECT COUNT(o) FROM Order o WHERE o.userId = :userId AND o.status = :status")
    long countByUserIdAndStatus(@Param("userId") Long userId, @Param("status") OrderStatus status);
}

// 3. 产品服务数据访问层
@Repository
public class ProductRepository extends JpaRepository {
    
    @Query("SELECT p FROM Product p WHERE p.category = :category AND p.status = 'ACTIVE'")
    List findByCategoryAndStatusActive(@Param("category") String category);
    
    @Query("SELECT p FROM Product p WHERE p.name LIKE %:keyword% OR p.description LIKE %:keyword%")
    List findByNameOrDescriptionContaining(@Param("keyword") String keyword);
    
    @Query("SELECT p FROM Product p WHERE p.price BETWEEN :minPrice AND :maxPrice")
    List findByPriceBetween(@Param("minPrice") BigDecimal minPrice, 
                                    @Param("maxPrice") BigDecimal maxPrice);
}

2. 分布式事务管理

// 1. Seata分布式事务配置
@Configuration
@EnableTransactionManagement
public class SeataConfig {
    
    @Bean
    public DataSourceProxy dataSourceProxy(DataSource dataSource) {
        return new DataSourceProxy(dataSource);
    }
    
    @Bean
    public SqlSessionFactory sqlSessionFactory(DataSourceProxy dataSourceProxy) throws Exception {
        SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
        factoryBean.setDataSource(dataSourceProxy);
        return factoryBean.getObject();
    }
}

// 2. 分布式事务服务
@Service
@Transactional
public class DistributedTransactionService {
    
    @Autowired
    private UserService userService;
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private InventoryService inventoryService;
    
    @GlobalTransactional(rollbackFor = Exception.class)
    public void createOrderWithUser(CreateOrderWithUserRequest request) {
        try {
            // 1. 创建用户
            UserDTO user = userService.createUser(request.getUserRequest());
            
            // 2. 创建订单
            CreateOrderRequest orderRequest = CreateOrderRequest.builder()
                .userId(user.getId())
                .items(request.getOrderItems())
                .build();
            OrderDTO order = orderService.createOrder(orderRequest);
            
            // 3. 扣减库存
            for (CreateOrderItemRequest item : request.getOrderItems()) {
                inventoryService.decreaseStock(item.getProductId(), item.getQuantity());
            }
            
            // 4. 发送通知
            notificationService.sendOrderConfirmation(user.getEmail(), order.getOrderNumber());
            
        } catch (Exception e) {
            log.error("Failed to create order with user", e);
            throw new BusinessException("Failed to create order with user", e);
        }
    }
}

// 3. 补偿事务
@Service
public class CompensationService {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private InventoryService inventoryService;
    
    @Autowired
    private NotificationService notificationService;
    
    @Transactional
    public void compensateOrderCreation(Long orderId, String reason) {
        try {
            // 1. 取消订单
            orderService.cancelOrder(orderId, reason);
            
            // 2. 恢复库存
            OrderDTO order = orderService.getOrderById(orderId);
            for (OrderItemDTO item : order.getItems()) {
                inventoryService.increaseStock(item.getProductId(), item.getQuantity());
            }
            
            // 3. 发送取消通知
            notificationService.sendOrderCancellation(order.getUserEmail(), order.getOrderNumber(), reason);
            
        } catch (Exception e) {
            log.error("Failed to compensate order creation for order: {}", orderId, e);
            // 记录补偿失败，需要人工处理
            auditService.logCompensationFailure(orderId, reason, e.getMessage());
        }
    }
}

六、微服务监控与治理

1. 链路追踪

// 1. Sleuth链路追踪配置
@Configuration
@EnableSleuth
public class SleuthConfig {
    
    @Bean
    public Sampler alwaysSampler() {
        return Sampler.ALWAYS_SAMPLE;
    }
    
    @Bean
    public SpanReporter spanReporter() {
        return new ZipkinSpanReporter();
    }
}

// 2. 自定义Span
@Service
public class TracingService {
    
    @Autowired
    private Tracer tracer;
    
    public void processOrder(Long orderId) {
        Span span = tracer.nextSpan()
            .name("process-order")
            .tag("order.id", orderId.toString())
            .start();
        
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            // 业务逻辑
            validateOrder(orderId);
            processPayment(orderId);
            updateInventory(orderId);
            sendNotification(orderId);
            
        } finally {
            span.end();
        }
    }
    
    private void validateOrder(Long orderId) {
        Span span = tracer.nextSpan()
            .name("validate-order")
            .tag("order.id", orderId.toString())
            .start();
        
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            // 验证订单逻辑
            Thread.sleep(100); // 模拟处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            span.end();
        }
    }
    
    private void processPayment(Long orderId) {
        Span span = tracer.nextSpan()
            .name("process-payment")
            .tag("order.id", orderId.toString())
            .start();
        
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            // 处理支付逻辑
            Thread.sleep(200); // 模拟处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            span.end();
        }
    }
    
    private void updateInventory(Long orderId) {
        Span span = tracer.nextSpan()
            .name("update-inventory")
            .tag("order.id", orderId.toString())
            .start();
        
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            // 更新库存逻辑
            Thread.sleep(150); // 模拟处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            span.end();
        }
    }
    
    private void sendNotification(Long orderId) {
        Span span = tracer.nextSpan()
            .name("send-notification")
            .tag("order.id", orderId.toString())
            .start();
        
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            // 发送通知逻辑
            Thread.sleep(50); // 模拟处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            span.end();
        }
    }
}

2. 服务治理

// 1. 服务健康检查
@Component
public class ServiceHealthIndicator implements HealthIndicator {
    
    @Autowired
    private DataSource dataSource;
    
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Override
    public Health health() {
        Health.Builder builder = Health.up();
        
        // 检查数据库连接
        try (Connection connection = dataSource.getConnection()) {
            if (connection.isValid(5)) {
                builder.withDetail("database", "UP");
            } else {
                builder.down().withDetail("database", "DOWN");
            }
        } catch (Exception e) {
            builder.down().withDetail("database", "DOWN").withDetail("error", e.getMessage());
        }
        
        // 检查Redis连接
        try {
            redisTemplate.opsForValue().get("health-check");
            builder.withDetail("redis", "UP");
        } catch (Exception e) {
            builder.down().withDetail("redis", "DOWN").withDetail("error", e.getMessage());
        }
        
        return builder.build();
    }
}

// 2. 服务降级
@Service
public class ServiceDegradationService {
    
    @Autowired
    private OrderServiceClient orderServiceClient;
    
    @Autowired
    private CacheService cacheService;
    
    @HystrixCommand(fallbackMethod = "getUserOrdersFallback")
    public List getUserOrders(Long userId) {
        return orderServiceClient.getOrdersByUserId(userId);
    }
    
    public List getUserOrdersFallback(Long userId) {
        log.warn("Order service is unavailable, returning cached data for user: {}", userId);
        
        // 返回缓存数据
        List cachedOrders = cacheService.getUserOrders(userId);
        if (cachedOrders != null) {
            return cachedOrders;
        }
        
        // 返回默认数据
        return Collections.emptyList();
    }
    
    @HystrixCommand(fallbackMethod = "createOrderFallback")
    public OrderDTO createOrder(CreateOrderRequest request) {
        return orderServiceClient.createOrder(request);
    }
    
    public OrderDTO createOrderFallback(CreateOrderRequest request) {
        log.error("Order service is unavailable, cannot create order for user: {}", request.getUserId());
        throw new ServiceUnavailableException("Order service is currently unavailable");
    }
}

// 3. 服务限流
@Component
public class RateLimitService {
    
    private final RedisTemplate redisTemplate;
    private final RateLimiter rateLimiter;
    
    public RateLimitService(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.rateLimiter = RateLimiter.create(100.0); // 每秒100个请求
    }
    
    public boolean isAllowed(String key, int limit, Duration window) {
        String redisKey = "rate_limit:" + key;
        String count = redisTemplate.opsForValue().get(redisKey);
        
        if (count == null) {
            redisTemplate.opsForValue().set(redisKey, "1", window);
            return true;
        }
        
        int currentCount = Integer.parseInt(count);
        if (currentCount >= limit) {
            return false;
        }
        
        redisTemplate.opsForValue().increment(redisKey);
        return true;
    }
    
    public boolean isAllowed(String key) {
        return isAllowed(key, 100, Duration.ofMinutes(1));
    }
}

七、微服务安全

1. JWT认证

// 1. JWT工具类
@Component
public class JwtTokenProvider {
    
    @Value("${jwt.secret}")
    private String secret;
    
    @Value("${jwt.expiration}")
    private long expiration;
    
    public String generateToken(UserDetails userDetails) {
        Map claims = new HashMap<>();
        claims.put("username", userDetails.getUsername());
        claims.put("authorities", userDetails.getAuthorities());
        
        return createToken(claims, userDetails.getUsername());
    }
    
    private String createToken(Map claims, String subject) {
        return Jwts.builder()
            .setClaims(claims)
            .setSubject(subject)
            .setIssuedAt(new Date())
            .setExpiration(new Date(System.currentTimeMillis() + expiration))
            .signWith(SignatureAlgorithm.HS512, secret)
            .compact();
    }
    
    public Boolean validateToken(String token, UserDetails userDetails) {
        final String username = getUsernameFromToken(token);
        return (username.equals(userDetails.getUsername()) && !isTokenExpired(token));
    }
    
    public String getUsernameFromToken(String token) {
        return getClaimFromToken(token, Claims::getSubject);
    }
    
    public Date getExpirationDateFromToken(String token) {
        return getClaimFromToken(token, Claims::getExpiration);
    }
    
    public  T getClaimFromToken(String token, Function claimsResolver) {
        final Claims claims = getAllClaimsFromToken(token);
        return claimsResolver.apply(claims);
    }
    
    private Claims getAllClaimsFromToken(String token) {
        return Jwts.parser().setSigningKey(secret).parseClaimsJws(token).getBody();
    }
    
    private Boolean isTokenExpired(String token) {
        final Date expiration = getExpirationDateFromToken(token);
        return expiration.before(new Date());
    }
}

// 2. JWT认证过滤器
@Component
public class JwtAuthenticationFilter extends OncePerRequestFilter {
    
    @Autowired
    private JwtTokenProvider jwtTokenProvider;
    
    @Autowired
    private UserDetailsService userDetailsService;
    
    @Override
    protected void doFilterInternal(HttpServletRequest request, 
                                  HttpServletResponse response, 
                                  FilterChain filterChain) throws ServletException, IOException {
        
        String token = getTokenFromRequest(request);
        
        if (token != null && jwtTokenProvider.validateToken(token, null)) {
            String username = jwtTokenProvider.getUsernameFromToken(token);
            UserDetails userDetails = userDetailsService.loadUserByUsername(username);
            
            UsernamePasswordAuthenticationToken authentication = 
                new UsernamePasswordAuthenticationToken(userDetails, null, userDetails.getAuthorities());
            SecurityContextHolder.getContext().setAuthentication(authentication);
        }
        
        filterChain.doFilter(request, response);
    }
    
    private String getTokenFromRequest(HttpServletRequest request) {
        String bearerToken = request.getHeader("Authorization");
        if (bearerToken != null && bearerToken.startsWith("Bearer ")) {
            return bearerToken.substring(7);
        }
        return null;
    }
}

2. 服务间认证

// 1. 服务间认证配置
@Configuration
@EnableWebSecurity
public class ServiceSecurityConfig {
    
    @Bean
    public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
        http
            .csrf().disable()
            .sessionManagement().sessionCreationPolicy(SessionCreationPolicy.STATELESS)
            .and()
            .authorizeRequests()
            .antMatchers("/api/public/**").permitAll()
            .antMatchers("/api/auth/**").permitAll()
            .antMatchers("/actuator/**").permitAll()
            .anyRequest().authenticated()
            .and()
            .addFilterBefore(new JwtAuthenticationFilter(), UsernamePasswordAuthenticationFilter.class);
        
        return http.build();
    }
    
    @Bean
    public PasswordEncoder passwordEncoder() {
        return new BCryptPasswordEncoder();
    }
}

// 2. 服务间调用认证
@Component
public class ServiceToServiceAuth {
    
    @Value("${service.auth.client-id}")
    private String clientId;
    
    @Value("${service.auth.client-secret}")
    private String clientSecret;
    
    @Autowired
    private RestTemplate restTemplate;
    
    public String getServiceToken() {
        try {
            String tokenUrl = "http://auth-service/oauth/token";
            
            MultiValueMap params = new LinkedMultiValueMap<>();
            params.add("grant_type", "client_credentials");
            params.add("client_id", clientId);
            params.add("client_secret", clientSecret);
            
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
            
            HttpEntity> request = new HttpEntity<>(params, headers);
            
            ResponseEntity

八、综合实战练习

1. 微服务项目结构

// 1. 项目结构示例
/*
microservices-project/
├── gateway-service/                 # API网关服务
│   ├── src/main/java/
│   │   └── com/example/gateway/
│   │       ├── GatewayApplication.java
│   │       ├── config/
│   │       │   ├── GatewayConfig.java
│   │       │   └── SecurityConfig.java
│   │       └── filter/
│   │           ├── AuthenticationFilter.java
│   │           └── RateLimitFilter.java
│   └── src/main/resources/
│       └── application.yml
├── user-service/                   # 用户服务
│   ├── src/main/java/
│   │   └── com/example/user/
│   │       ├── UserApplication.java
│   │       ├── controller/
│   │       │   └── UserController.java
│   │       ├── service/
│   │       │   └── UserService.java
│   │       ├── repository/
│   │       │   └── UserRepository.java
│   │       └── model/
│   │           ├── User.java
│   │           └── UserDTO.java
│   └── src/main/resources/
│       └── application.yml
├── order-service/                  # 订单服务
│   ├── src/main/java/
│   │   └── com/example/order/
│   │       ├── OrderApplication.java
│   │       ├── controller/
│   │       │   └── OrderController.java
│   │       ├── service/
│   │       │   └── OrderService.java
│   │       ├── repository/
│   │       │   └── OrderRepository.java
│   │       └── model/
│   │           ├── Order.java
│   │           └── OrderDTO.java
│   └── src/main/resources/
│       └── application.yml
└── product-service/                # 产品服务
    ├── src/main/java/
    │   └── com/example/product/
    │       ├── ProductApplication.java
    │       ├── controller/
    │       │   └── ProductController.java
    │       ├── service/
    │       │   └── ProductService.java
    │       ├── repository/
    │       │   └── ProductRepository.java
    │       └── model/
    │           ├── Product.java
    │           └── ProductDTO.java
    └── src/main/resources/
        └── application.yml
*/

// 2. 服务配置示例
@SpringBootApplication
@EnableEurekaClient
@EnableDiscoveryClient
@EnableFeignClients
public class UserServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(UserServiceApplication.class, args);
    }
}

// 3. 服务间调用示例
@FeignClient(name = "order-service", path = "/api/orders")
public interface OrderServiceClient {
    
    @GetMapping("/user/{userId}")
    List getOrdersByUserId(@PathVariable("userId") Long userId);
    
    @PostMapping
    OrderDTO createOrder(@RequestBody CreateOrderRequest request);
    
    @GetMapping("/{orderId}")
    OrderDTO getOrderById(@PathVariable("orderId") Long orderId);
}

2. 微服务部署配置

# docker-compose.yml
version: '3.8'

services:
  eureka-server:
    image: eureka-server:latest
    ports:
      - "8761:8761"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
    networks:
      - microservices-network

  gateway-service:
    image: gateway-service:latest
    ports:
      - "8080:8080"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
      - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka
    depends_on:
      - eureka-server
    networks:
      - microservices-network

  user-service:
    image: user-service:latest
    ports:
      - "8081:8081"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
      - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka
      - SPRING_DATASOURCE_URL=jdbc:postgresql://postgres:5432/userdb
    depends_on:
      - eureka-server
      - postgres
    networks:
      - microservices-network

  order-service:
    image: order-service:latest
    ports:
      - "8082:8082"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
      - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka
      - SPRING_DATASOURCE_URL=jdbc:postgresql://postgres:5432/orderdb
    depends_on:
      - eureka-server
      - postgres
    networks:
      - microservices-network

  product-service:
    image: product-service:latest
    ports:
      - "8083:8083"
    environment:
      - SPRING_PROFILES_ACTIVE