Java微服务架构设计与实践

微服务架构作为现代软件开发的主流架构模式，通过将大型应用拆分为多个小型、独立的服务，实现了更好的可扩展性、可维护性和技术多样性。本文将深入探讨Java微服务架构的设计原则、技术实现和最佳实践。

微服务架构概述

微服务架构的核心特征

1. 服务独立性：每个服务独立开发、部署和扩展
2. 业务导向：围绕业务能力组织服务
3. 去中心化：数据管理和治理去中心化
4. 容错性：设计时考虑故障处理
5. 技术多样性：不同服务可使用不同技术栈

微服务 vs 单体架构

// 单体架构示例
@RestController
public class MonolithController {
    
    @Autowired
    private UserService userService;
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private PaymentService paymentService;
    
    @PostMapping("/orders")
    public ResponseEntity<Order> createOrder(@RequestBody OrderRequest request) {
        // 所有业务逻辑在一个应用中
        User user = userService.getUser(request.getUserId());
        Order order = orderService.createOrder(user, request.getItems());
        Payment payment = paymentService.processPayment(order);
        
        return ResponseEntity.ok(order);
    }
}

// 微服务架构示例
// 用户服务
@RestController
@RequestMapping("/api/users")
public class UserController {
    
    @Autowired
    private UserService userService;
    
    @GetMapping("/{userId}")
    public ResponseEntity<User> getUser(@PathVariable Long userId) {
        User user = userService.findById(userId);
        return ResponseEntity.ok(user);
    }
}

// 订单服务
@RestController
@RequestMapping("/api/orders")
public class OrderController {
    
    @Autowired
    private OrderService orderService;
    
    @Autowired
    private UserServiceClient userServiceClient;
    
    @PostMapping
    public ResponseEntity<Order> createOrder(@RequestBody OrderRequest request) {
        // 通过服务调用获取用户信息
        User user = userServiceClient.getUser(request.getUserId());
        Order order = orderService.createOrder(user, request.getItems());
        
        return ResponseEntity.ok(order);
    }
}

Spring Boot微服务基础

创建微服务项目结构

// 主应用类
@SpringBootApplication
@EnableEurekaClient
@EnableFeignClients
@EnableCircuitBreaker
public class UserServiceApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(UserServiceApplication.class, args);
    }
    
    @Bean
    @LoadBalanced
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }
}

// 用户实体
@Entity
@Table(name = "users")
public class User {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(nullable = false, unique = true)
    private String username;
    
    @Column(nullable = false)
    private String email;
    
    @Column(nullable = false)
    private String password;
    
    @CreationTimestamp
    private LocalDateTime createdAt;
    
    @UpdateTimestamp
    private LocalDateTime updatedAt;
    
    // 构造函数、getter、setter
    public User() {}
    
    public User(String username, String email, String password) {
        this.username = username;
        this.email = email;
        this.password = password;
    }
    
    // getters and setters
    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }
    
    public String getUsername() { return username; }
    public void setUsername(String username) { this.username = username; }
    
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
    
    public String getPassword() { return password; }
    public void setPassword(String password) { this.password = password; }
    
    public LocalDateTime getCreatedAt() { return createdAt; }
    public void setCreatedAt(LocalDateTime createdAt) { this.createdAt = createdAt; }
    
    public LocalDateTime getUpdatedAt() { return updatedAt; }
    public void setUpdatedAt(LocalDateTime updatedAt) { this.updatedAt = updatedAt; }
}

// 用户仓库
@Repository
public interface UserRepository extends JpaRepository<User, Long> {
    
    Optional<User> findByUsername(String username);
    
    Optional<User> findByEmail(String email);
    
    @Query("SELECT u FROM User u WHERE u.createdAt >= :startDate")
    List<User> findUsersCreatedAfter(@Param("startDate") LocalDateTime startDate);
    
    @Modifying
    @Query("UPDATE User u SET u.email = :email WHERE u.id = :id")
    int updateUserEmail(@Param("id") Long id, @Param("email") String email);
}

// 用户服务
@Service
@Transactional
public class UserService {
    
    private final UserRepository userRepository;
    private final PasswordEncoder passwordEncoder;
    private final ApplicationEventPublisher eventPublisher;
    
    public UserService(UserRepository userRepository, 
                      PasswordEncoder passwordEncoder,
                      ApplicationEventPublisher eventPublisher) {
        this.userRepository = userRepository;
        this.passwordEncoder = passwordEncoder;
        this.eventPublisher = eventPublisher;
    }
    
    public User createUser(CreateUserRequest request) {
        // 验证用户名和邮箱唯一性
        if (userRepository.findByUsername(request.getUsername()).isPresent()) {
            throw new UserAlreadyExistsException("用户名已存在: " + request.getUsername());
        }
        
        if (userRepository.findByEmail(request.getEmail()).isPresent()) {
            throw new UserAlreadyExistsException("邮箱已存在: " + request.getEmail());
        }
        
        // 创建用户
        User user = new User(
            request.getUsername(),
            request.getEmail(),
            passwordEncoder.encode(request.getPassword())
        );
        
        User savedUser = userRepository.save(user);
        
        // 发布用户创建事件
        eventPublisher.publishEvent(new UserCreatedEvent(savedUser));
        
        return savedUser;
    }
    
    @Transactional(readOnly = true)
    public User getUserById(Long id) {
        return userRepository.findById(id)
            .orElseThrow(() -> new UserNotFoundException("用户不存在: " + id));
    }
    
    @Transactional(readOnly = true)
    public User getUserByUsername(String username) {
        return userRepository.findByUsername(username)
            .orElseThrow(() -> new UserNotFoundException("用户不存在: " + username));
    }
    
    public User updateUser(Long id, UpdateUserRequest request) {
        User user = getUserById(id);
        
        if (request.getEmail() != null && !request.getEmail().equals(user.getEmail())) {
            if (userRepository.findByEmail(request.getEmail()).isPresent()) {
                throw new UserAlreadyExistsException("邮箱已存在: " + request.getEmail());
            }
            user.setEmail(request.getEmail());
        }
        
        User updatedUser = userRepository.save(user);
        
        // 发布用户更新事件
        eventPublisher.publishEvent(new UserUpdatedEvent(updatedUser));
        
        return updatedUser;
    }
    
    public void deleteUser(Long id) {
        User user = getUserById(id);
        userRepository.delete(user);
        
        // 发布用户删除事件
        eventPublisher.publishEvent(new UserDeletedEvent(user));
    }
    
    @Transactional(readOnly = true)
    public Page<User> getUsers(Pageable pageable) {
        return userRepository.findAll(pageable);
    }
}

// 用户控制器
@RestController
@RequestMapping("/api/users")
@Validated
public class UserController {
    
    private final UserService userService;
    
    public UserController(UserService userService) {
        this.userService = userService;
    }
    
    @PostMapping
    public ResponseEntity<UserResponse> createUser(@Valid @RequestBody CreateUserRequest request) {
        User user = userService.createUser(request);
        UserResponse response = UserResponse.from(user);
        
        return ResponseEntity.status(HttpStatus.CREATED)
            .location(URI.create("/api/users/" + user.getId()))
            .body(response);
    }
    
    @GetMapping("/{id}")
    public ResponseEntity<UserResponse> getUser(@PathVariable Long id) {
        User user = userService.getUserById(id);
        UserResponse response = UserResponse.from(user);
        
        return ResponseEntity.ok(response);
    }
    
    @GetMapping
    public ResponseEntity<Page<UserResponse>> getUsers(
            @PageableDefault(size = 20, sort = "createdAt", direction = Sort.Direction.DESC) 
            Pageable pageable) {
        
        Page<User> users = userService.getUsers(pageable);
        Page<UserResponse> response = users.map(UserResponse::from);
        
        return ResponseEntity.ok(response);
    }
    
    @PutMapping("/{id}")
    public ResponseEntity<UserResponse> updateUser(
            @PathVariable Long id, 
            @Valid @RequestBody UpdateUserRequest request) {
        
        User user = userService.updateUser(id, request);
        UserResponse response = UserResponse.from(user);
        
        return ResponseEntity.ok(response);
    }
    
    @DeleteMapping("/{id}")
    public ResponseEntity<Void> deleteUser(@PathVariable Long id) {
        userService.deleteUser(id);
        return ResponseEntity.noContent().build();
    }
}

// 请求和响应DTO
public class CreateUserRequest {
    @NotBlank(message = "用户名不能为空")
    @Size(min = 3, max = 50, message = "用户名长度必须在3-50之间")
    private String username;
    
    @NotBlank(message = "邮箱不能为空")
    @Email(message = "邮箱格式不正确")
    private String email;
    
    @NotBlank(message = "密码不能为空")
    @Size(min = 6, max = 100, message = "密码长度必须在6-100之间")
    private String password;
    
    // getters and setters
    public String getUsername() { return username; }
    public void setUsername(String username) { this.username = username; }
    
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
    
    public String getPassword() { return password; }
    public void setPassword(String password) { this.password = password; }
}

public class UpdateUserRequest {
    @Email(message = "邮箱格式不正确")
    private String email;
    
    // getters and setters
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
}

public class UserResponse {
    private Long id;
    private String username;
    private String email;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
    
    public static UserResponse from(User user) {
        UserResponse response = new UserResponse();
        response.setId(user.getId());
        response.setUsername(user.getUsername());
        response.setEmail(user.getEmail());
        response.setCreatedAt(user.getCreatedAt());
        response.setUpdatedAt(user.getUpdatedAt());
        return response;
    }
    
    // getters and setters
    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }
    
    public String getUsername() { return username; }
    public void setUsername(String username) { this.username = username; }
    
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
    
    public LocalDateTime getCreatedAt() { return createdAt; }
    public void setCreatedAt(LocalDateTime createdAt) { this.createdAt = createdAt; }
    
    public LocalDateTime getUpdatedAt() { return updatedAt; }
    public void setUpdatedAt(LocalDateTime updatedAt) { this.updatedAt = updatedAt; }
}

Spring Cloud服务治理

服务注册与发现 - Eureka

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

// application.yml for Eureka Server
/*
server:
  port: 8761

eureka:
  instance:
    hostname: localhost
  client:
    register-with-eureka: false
    fetch-registry: false
    service-url:
      defaultZone: http://${eureka.instance.hostname}:${server.port}/eureka/
  server:
    enable-self-preservation: false
*/

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

// application.yml for Service Client
/*
server:
  port: 8081

spring:
  application:
    name: user-service
  datasource:
    url: jdbc:mysql://localhost:3306/user_service
    username: root
    password: password
    driver-class-name: com.mysql.cj.jdbc.Driver
  jpa:
    hibernate:
      ddl-auto: update
    show-sql: true

eureka:
  client:
    service-url:
      defaultZone: http://localhost:8761/eureka/
  instance:
    prefer-ip-address: true
    lease-renewal-interval-in-seconds: 30
    lease-expiration-duration-in-seconds: 90
*/

// 服务发现示例
@Component
public class ServiceDiscoveryExample {
    
    @Autowired
    private DiscoveryClient discoveryClient;
    
    public List<ServiceInstance> getServiceInstances(String serviceName) {
        return discoveryClient.getInstances(serviceName);
    }
    
    public List<String> getServices() {
        return discoveryClient.getServices();
    }
    
    public void printServiceInfo() {
        List<String> services = getServices();
        
        for (String service : services) {
            System.out.println("服务名称: " + service);
            List<ServiceInstance> instances = getServiceInstances(service);
            
            for (ServiceInstance instance : instances) {
                System.out.println("  实例ID: " + instance.getInstanceId());
                System.out.println("  主机: " + instance.getHost());
                System.out.println("  端口: " + instance.getPort());
                System.out.println("  URI: " + instance.getUri());
                System.out.println("  元数据: " + instance.getMetadata());
                System.out.println("  ---");
            }
        }
    }
}

服务间通信 - Feign

// Feign客户端接口
@FeignClient(name = "user-service", fallback = UserServiceFallback.class)
public interface UserServiceClient {
    
    @GetMapping("/api/users/{id}")
    UserResponse getUser(@PathVariable("id") Long id);
    
    @PostMapping("/api/users")
    UserResponse createUser(@RequestBody CreateUserRequest request);
    
    @GetMapping("/api/users")
    Page<UserResponse> getUsers(@RequestParam("page") int page, 
                               @RequestParam("size") int size);
    
    @PutMapping("/api/users/{id}")
    UserResponse updateUser(@PathVariable("id") Long id, 
                           @RequestBody UpdateUserRequest request);
    
    @DeleteMapping("/api/users/{id}")
    void deleteUser(@PathVariable("id") Long id);
}

// Feign客户端降级处理
@Component
public class UserServiceFallback implements UserServiceClient {
    
    @Override
    public UserResponse getUser(Long id) {
        UserResponse fallbackUser = new UserResponse();
        fallbackUser.setId(id);
        fallbackUser.setUsername("fallback-user");
        fallbackUser.setEmail("fallback@example.com");
        return fallbackUser;
    }
    
    @Override
    public UserResponse createUser(CreateUserRequest request) {
        throw new ServiceUnavailableException("用户服务暂时不可用，无法创建用户");
    }
    
    @Override
    public Page<UserResponse> getUsers(int page, int size) {
        return Page.empty();
    }
    
    @Override
    public UserResponse updateUser(Long id, UpdateUserRequest request) {
        throw new ServiceUnavailableException("用户服务暂时不可用，无法更新用户");
    }
    
    @Override
    public void deleteUser(Long id) {
        throw new ServiceUnavailableException("用户服务暂时不可用，无法删除用户");
    }
}

// Feign配置
@Configuration
public class FeignConfig {
    
    @Bean
    public Logger.Level feignLoggerLevel() {
        return Logger.Level.FULL;
    }
    
    @Bean
    public RequestInterceptor requestInterceptor() {
        return requestTemplate -> {
            // 添加认证头
            String token = getCurrentUserToken();
            if (token != null) {
                requestTemplate.header("Authorization", "Bearer " + token);
            }
            
            // 添加请求ID用于链路追踪
            String requestId = UUID.randomUUID().toString();
            requestTemplate.header("X-Request-ID", requestId);
        };
    }
    
    @Bean
    public ErrorDecoder errorDecoder() {
        return new CustomErrorDecoder();
    }
    
    private String getCurrentUserToken() {
        // 从当前上下文获取用户token
        return SecurityContextHolder.getContext()
            .getAuthentication()
            .getCredentials()
            .toString();
    }
}

// 自定义错误解码器
public class CustomErrorDecoder implements ErrorDecoder {
    
    private final ErrorDecoder defaultErrorDecoder = new Default();
    
    @Override
    public Exception decode(String methodKey, Response response) {
        switch (response.status()) {
            case 400:
                return new BadRequestException("请求参数错误");
            case 401:
                return new UnauthorizedException("未授权访问");
            case 403:
                return new ForbiddenException("访问被拒绝");
            case 404:
                return new NotFoundException("资源不存在");
            case 500:
                return new InternalServerErrorException("服务器内部错误");
            case 503:
                return new ServiceUnavailableException("服务暂时不可用");
            default:
                return defaultErrorDecoder.decode(methodKey, response);
        }
    }
}

// 在订单服务中使用用户服务客户端
@Service
public class OrderService {
    
    private final OrderRepository orderRepository;
    private final UserServiceClient userServiceClient;
    
    public OrderService(OrderRepository orderRepository, 
                       UserServiceClient userServiceClient) {
        this.orderRepository = orderRepository;
        this.userServiceClient = userServiceClient;
    }
    
    public Order createOrder(CreateOrderRequest request) {
        // 验证用户存在
        UserResponse user = userServiceClient.getUser(request.getUserId());
        
        // 创建订单
        Order order = new Order();
        order.setUserId(user.getId());
        order.setUserName(user.getUsername());
        order.setItems(request.getItems());
        order.setTotalAmount(calculateTotalAmount(request.getItems()));
        order.setStatus(OrderStatus.PENDING);
        
        return orderRepository.save(order);
    }
    
    private BigDecimal calculateTotalAmount(List<OrderItem> items) {
        return items.stream()
            .map(item -> item.getPrice().multiply(BigDecimal.valueOf(item.getQuantity())))
            .reduce(BigDecimal.ZERO, BigDecimal::add);
    }
}

负载均衡 - Ribbon

// Ribbon配置
@Configuration
public class RibbonConfig {
    
    @Bean
    @LoadBalanced
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }
    
    // 自定义负载均衡规则
    @Bean
    public IRule ribbonRule() {
        // 可选择不同的负载均衡策略
        return new WeightedResponseTimeRule(); // 响应时间加权
        // return new RoundRobinRule(); // 轮询
        // return new RandomRule(); // 随机
        // return new BestAvailableRule(); // 最少并发
    }
    
    // 自定义Ping策略
    @Bean
    public IPing ribbonPing() {
        return new PingUrl();
    }
    
    // 服务列表更新策略
    @Bean
    public ServerListUpdater ribbonServerListUpdater() {
        return new PollingServerListUpdater();
    }
}

// 使用RestTemplate进行服务调用
@Service
public class OrderServiceUsingRestTemplate {
    
    @Autowired
    @LoadBalanced
    private RestTemplate restTemplate;
    
    public UserResponse getUserById(Long userId) {
        String url = "http://user-service/api/users/" + userId;
        
        try {
            return restTemplate.getForObject(url, UserResponse.class);
        } catch (Exception e) {
            // 处理服务调用异常
            throw new ServiceCallException("调用用户服务失败", e);
        }
    }
    
    public UserResponse createUser(CreateUserRequest request) {
        String url = "http://user-service/api/users";
        
        try {
            return restTemplate.postForObject(url, request, UserResponse.class);
        } catch (Exception e) {
            throw new ServiceCallException("创建用户失败", e);
        }
    }
    
    // 带重试机制的服务调用
    @Retryable(value = {Exception.class}, maxAttempts = 3, backoff = @Backoff(delay = 1000))
    public UserResponse getUserWithRetry(Long userId) {
        return getUserById(userId);
    }
    
    @Recover
    public UserResponse recoverGetUser(Exception ex, Long userId) {
        // 重试失败后的恢复逻辑
        UserResponse fallbackUser = new UserResponse();
        fallbackUser.setId(userId);
        fallbackUser.setUsername("fallback-user");
        return fallbackUser;
    }
}

// 自定义负载均衡规则
public class CustomLoadBalancingRule extends AbstractLoadBalancerRule {
    
    @Override
    public Server choose(Object key) {
        ILoadBalancer lb = getLoadBalancer();
        if (lb == null) {
            return null;
        }
        
        List<Server> servers = lb.getReachableServers();
        if (servers.isEmpty()) {
            return null;
        }
        
        // 自定义选择逻辑：优先选择响应时间最短的服务器
        Server chosenServer = null;
        long minResponseTime = Long.MAX_VALUE;
        
        for (Server server : servers) {
            ServerStats stats = ((AbstractLoadBalancer) lb).getLoadBalancerStats()
                .getSingleServerStat(server);
            
            if (stats != null) {
                long avgResponseTime = (long) stats.getResponseTimeAvg();
                if (avgResponseTime < minResponseTime) {
                    minResponseTime = avgResponseTime;
                    chosenServer = server;
                }
            }
        }
        
        return chosenServer != null ? chosenServer : servers.get(0);
    }
    
    @Override
    public void initWithNiwsConfig(IClientConfig clientConfig) {
        // 初始化配置
    }
}

断路器模式 - Hystrix

// Hystrix配置
@Configuration
@EnableCircuitBreaker
public class HystrixConfig {
    
    @Bean
    public HystrixCommandAspect hystrixAspect() {
        return new HystrixCommandAspect();
    }
}

// 使用Hystrix的服务类
@Service
public class UserServiceWithHystrix {
    
    @Autowired
    private UserServiceClient userServiceClient;
    
    @HystrixCommand(
        fallbackMethod = "getUserFallback",
        commandProperties = {
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "3000"),
            @HystrixProperty(name = "circuitBreaker.requestVolumeThreshold", value = "10"),
            @HystrixProperty(name = "circuitBreaker.errorThresholdPercentage", value = "50"),
            @HystrixProperty(name = "circuitBreaker.sleepWindowInMilliseconds", value = "10000")
        },
        threadPoolProperties = {
            @HystrixProperty(name = "coreSize", value = "10"),
            @HystrixProperty(name = "maxQueueSize", value = "100")
        }
    )
    public UserResponse getUser(Long userId) {
        return userServiceClient.getUser(userId);
    }
    
    // 降级方法
    public UserResponse getUserFallback(Long userId) {
        UserResponse fallbackUser = new UserResponse();
        fallbackUser.setId(userId);
        fallbackUser.setUsername("fallback-user");
        fallbackUser.setEmail("fallback@example.com");
        return fallbackUser;
    }
    
    // 批量获取用户，使用请求合并
    @HystrixCollapser(
        batchMethod = "getUsersBatch",
        collapserProperties = {
            @HystrixProperty(name = "timerDelayInMilliseconds", value = "100"),
            @HystrixProperty(name = "maxRequestsInBatch", value = "10")
        }
    )
    public Future<UserResponse> getUserAsync(Long userId) {
        // 这个方法不会被直接执行，会被合并到批量方法中
        return null;
    }
    
    @HystrixCommand
    public List<UserResponse> getUsersBatch(List<Long> userIds) {
        // 批量获取用户信息
        return userIds.stream()
            .map(this::getUser)
            .collect(Collectors.toList());
    }
    
    // 缓存示例
    @HystrixCommand(
        commandKey = "getUserById",
        fallbackMethod = "getUserFallback"
    )
    @CacheResult
    public UserResponse getUserWithCache(@CacheKey Long userId) {
        return userServiceClient.getUser(userId);
    }
    
    @CacheRemove(commandKey = "getUserById")
    public void evictUserCache(@CacheKey Long userId) {
        // 清除缓存
    }
}

// Hystrix监控
@RestController
@RequestMapping("/hystrix")
public class HystrixMonitorController {
    
    @GetMapping("/stream")
    public void hystrixStream(HttpServletResponse response) throws IOException {
        response.setContentType("text/event-stream");
        response.setCharacterEncoding("UTF-8");
        response.setHeader("Cache-Control", "no-cache");
        
        // 输出Hystrix监控数据流
        // 实际实现中会使用HystrixMetricsStreamServlet
    }
}

// 自定义Hystrix命令
public class GetUserCommand extends HystrixCommand<UserResponse> {
    
    private final UserServiceClient userServiceClient;
    private final Long userId;
    
    public GetUserCommand(UserServiceClient userServiceClient, Long userId) {
        super(HystrixCommandGroupKey.Factory.asKey("UserService"));
        this.userServiceClient = userServiceClient;
        this.userId = userId;
    }
    
    @Override
    protected UserResponse run() throws Exception {
        return userServiceClient.getUser(userId);
    }
    
    @Override
    protected UserResponse getFallback() {
        UserResponse fallbackUser = new UserResponse();
        fallbackUser.setId(userId);
        fallbackUser.setUsername("fallback-user");
        return fallbackUser;
    }
    
    @Override
    protected String getCacheKey() {
        return "user-" + userId;
    }
}

// 使用自定义命令
@Service
public class OrderServiceWithCustomCommand {
    
    private final UserServiceClient userServiceClient;
    
    public OrderServiceWithCustomCommand(UserServiceClient userServiceClient) {
        this.userServiceClient = userServiceClient;
    }
    
    public UserResponse getUser(Long userId) {
        return new GetUserCommand(userServiceClient, userId).execute();
    }
    
    public Future<UserResponse> getUserAsync(Long userId) {
        return new GetUserCommand(userServiceClient, userId).queue();
    }
    
    public Observable<UserResponse> getUserObservable(Long userId) {
        return new GetUserCommand(userServiceClient, userId).observe();
    }
}

API网关 - Spring Cloud Gateway

// Gateway配置
@SpringBootApplication
public class GatewayApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(GatewayApplication.class, args);
    }
    
    @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-cb")
                        .setFallbackUri("forward:/fallback/users"))
                    .retry(config -> config
                        .setRetries(3)
                        .setBackoff(Duration.ofMillis(100), Duration.ofMillis(1000), 2, true)))
                .uri("lb://user-service"))
            
            // 订单服务路由
            .route("order-service", r -> r
                .path("/api/orders/**")
                .filters(f -> f
                    .addRequestHeader("X-Gateway", "Spring-Cloud-Gateway")
                    .requestRateLimiter(config -> config
                        .setRateLimiter(redisRateLimiter())
                        .setKeyResolver(userKeyResolver())))
                .uri("lb://order-service"))
            
            // 支付服务路由（需要认证）
            .route("payment-service", r -> r
                .path("/api/payments/**")
                .filters(f -> f
                    .addRequestHeader("X-Gateway", "Spring-Cloud-Gateway")
                    .filter(authenticationFilter()))
                .uri("lb://payment-service"))
            
            .build();
    }
    
    // Redis限流器
    @Bean
    public RedisRateLimiter redisRateLimiter() {
        return new RedisRateLimiter(10, 20); // 每秒10个请求，突发20个
    }
    
    // 用户Key解析器
    @Bean
    public KeyResolver userKeyResolver() {
        return exchange -> {
            String userId = exchange.getRequest().getHeaders().getFirst("X-User-ID");
            return Mono.just(userId != null ? userId : "anonymous");
        };
    }
    
    // 认证过滤器
    @Bean
    public GatewayFilter authenticationFilter() {
        return (exchange, chain) -> {
            ServerHttpRequest request = exchange.getRequest();
            String authHeader = request.getHeaders().getFirst("Authorization");
            
            if (authHeader == null || !authHeader.startsWith("Bearer ")) {
                ServerHttpResponse response = exchange.getResponse();
                response.setStatusCode(HttpStatus.UNAUTHORIZED);
                return response.setComplete();
            }
            
            // 验证JWT token
            String token = authHeader.substring(7);
            if (!isValidToken(token)) {
                ServerHttpResponse response = exchange.getResponse();
                response.setStatusCode(HttpStatus.UNAUTHORIZED);
                return response.setComplete();
            }
            
            return chain.filter(exchange);
        };
    }
    
    private boolean isValidToken(String token) {
        // 实现JWT token验证逻辑
        try {
            // 解析和验证token
            return true;
        } catch (Exception e) {
            return false;
        }
    }
}

// 全局过滤器
@Component
public class GlobalLoggingFilter implements GlobalFilter, Ordered {
    
    private static final Logger logger = LoggerFactory.getLogger(GlobalLoggingFilter.class);
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        
        // 记录请求信息
        logger.info("请求路径: {}, 方法: {}, 参数: {}", 
            request.getPath(), 
            request.getMethod(), 
            request.getQueryParams());
        
        // 添加请求ID
        String requestId = UUID.randomUUID().toString();
        ServerHttpRequest mutatedRequest = request.mutate()
            .header("X-Request-ID", requestId)
            .build();
        
        ServerWebExchange mutatedExchange = exchange.mutate()
            .request(mutatedRequest)
            .build();
        
        long startTime = System.currentTimeMillis();
        
        return chain.filter(mutatedExchange).then(Mono.fromRunnable(() -> {
            long endTime = System.currentTimeMillis();
            ServerHttpResponse response = exchange.getResponse();
            
            logger.info("请求完成 - ID: {}, 状态码: {}, 耗时: {}ms", 
                requestId, 
                response.getStatusCode(), 
                endTime - startTime);
        }));
    }
    
    @Override
    public int getOrder() {
        return -1; // 最高优先级
    }
}

// 降级处理
@RestController
public class FallbackController {
    
    @RequestMapping("/fallback/users")
    public Mono<ResponseEntity<String>> userFallback() {
        return Mono.just(ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
            .body("用户服务暂时不可用，请稍后重试"));
    }
    
    @RequestMapping("/fallback/orders")
    public Mono<ResponseEntity<String>> orderFallback() {
        return Mono.just(ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE)
            .body("订单服务暂时不可用，请稍后重试"));
    }
}

// 自定义过滤器工厂
@Component
public class CustomGatewayFilterFactory extends AbstractGatewayFilterFactory<CustomGatewayFilterFactory.Config> {
    
    public CustomGatewayFilterFactory() {
        super(Config.class);
    }
    
    @Override
    public GatewayFilter apply(Config config) {
        return (exchange, chain) -> {
            ServerHttpRequest request = exchange.getRequest();
            
            // 添加自定义逻辑
            if (config.isPreLogger()) {
                System.out.println("Pre Filter: " + request.getURI());
            }
            
            return chain.filter(exchange).then(Mono.fromRunnable(() -> {
                if (config.isPostLogger()) {
                    System.out.println("Post Filter: " + exchange.getResponse().getStatusCode());
                }
            }));
        };
    }
    
    public static class Config {
        private boolean preLogger;
        private boolean postLogger;
        
        // getters and setters
        public boolean isPreLogger() { return preLogger; }
        public void setPreLogger(boolean preLogger) { this.preLogger = preLogger; }
        
        public boolean isPostLogger() { return postLogger; }
        public void setPostLogger(boolean postLogger) { this.postLogger = postLogger; }
    }
}

配置管理 - Spring Cloud Config

// Config Server
@SpringBootApplication
@EnableConfigServer
public class ConfigServerApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(ConfigServerApplication.class, args);
    }
}

// Config Server配置 (application.yml)
/*
server:
  port: 8888

spring:
  application:
    name: config-server
  cloud:
    config:
      server:
        git:
          uri: https://github.com/your-org/config-repo
          search-paths: '{application}'
          default-label: main
        health:
          repositories:
            user-service:
              label: main
              name: user-service
              profiles: dev,prod
*/

// Config Client配置
@SpringBootApplication
@EnableEurekaClient
public class UserServiceApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(UserServiceApplication.class, args);
    }
}

// bootstrap.yml for Config Client
/*
spring:
  application:
    name: user-service
  cloud:
    config:
      uri: http://localhost:8888
      profile: dev
      label: main
      fail-fast: true
      retry:
        initial-interval: 1000
        max-attempts: 6
        max-interval: 2000
        multiplier: 1.1
*/

// 配置属性类
@ConfigurationProperties(prefix = "user-service")
@RefreshScope
@Component
public class UserServiceConfig {
    
    private String welcomeMessage;
    private int maxUsersPerPage;
    private boolean enableUserRegistration;
    private Database database;
    private Security security;
    
    public static class Database {
        private int maxConnections;
        private int connectionTimeout;
        private String url;
        
        // getters and setters
        public int getMaxConnections() { return maxConnections; }
        public void setMaxConnections(int maxConnections) { this.maxConnections = maxConnections; }
        
        public int getConnectionTimeout() { return connectionTimeout; }
        public void setConnectionTimeout(int connectionTimeout) { this.connectionTimeout = connectionTimeout; }
        
        public String getUrl() { return url; }
        public void setUrl(String url) { this.url = url; }
    }
    
    public static class Security {
        private String jwtSecret;
        private int jwtExpirationHours;
        
        // getters and setters
        public String getJwtSecret() { return jwtSecret; }
        public void setJwtSecret(String jwtSecret) { this.jwtSecret = jwtSecret; }
        
        public int getJwtExpirationHours() { return jwtExpirationHours; }
        public void setJwtExpirationHours(int jwtExpirationHours) { this.jwtExpirationHours = jwtExpirationHours; }
    }
    
    // getters and setters
    public String getWelcomeMessage() { return welcomeMessage; }
    public void setWelcomeMessage(String welcomeMessage) { this.welcomeMessage = welcomeMessage; }
    
    public int getMaxUsersPerPage() { return maxUsersPerPage; }
    public void setMaxUsersPerPage(int maxUsersPerPage) { this.maxUsersPerPage = maxUsersPerPage; }
    
    public boolean isEnableUserRegistration() { return enableUserRegistration; }
    public void setEnableUserRegistration(boolean enableUserRegistration) { this.enableUserRegistration = enableUserRegistration; }
    
    public Database getDatabase() { return database; }
    public void setDatabase(Database database) { this.database = database; }
    
    public Security getSecurity() { return security; }
    public void setSecurity(Security security) { this.security = security; }
}

// 使用配置的服务
@Service
@RefreshScope
public class ConfigurableUserService {
    
    private final UserServiceConfig config;
    private final UserRepository userRepository;
    
    public ConfigurableUserService(UserServiceConfig config, UserRepository userRepository) {
        this.config = config;
        this.userRepository = userRepository;
    }
    
    public String getWelcomeMessage() {
        return config.getWelcomeMessage();
    }
    
    public Page<User> getUsers(Pageable pageable) {
        // 使用配置中的最大页面大小
        int maxSize = config.getMaxUsersPerPage();
        if (pageable.getPageSize() > maxSize) {
            pageable = PageRequest.of(
                pageable.getPageNumber(), 
                maxSize, 
                pageable.getSort()
            );
        }
        
        return userRepository.findAll(pageable);
    }
    
    public User createUser(CreateUserRequest request) {
        if (!config.isEnableUserRegistration()) {
            throw new UserRegistrationDisabledException("用户注册功能已禁用");
        }
        
        // 创建用户逻辑
        return new User(request.getUsername(), request.getEmail(), request.getPassword());
    }
}

// 配置刷新端点
@RestController
@RequestMapping("/config")
public class ConfigController {
    
    private final UserServiceConfig config;
    
    public ConfigController(UserServiceConfig config) {
        this.config = config;
    }
    
    @GetMapping("/current")
    public ResponseEntity<UserServiceConfig> getCurrentConfig() {
        return ResponseEntity.ok(config);
    }
    
    @PostMapping("/refresh")
    public ResponseEntity<String> refreshConfig() {
        // 手动触发配置刷新
        // 实际上Spring Cloud Bus可以自动处理
        return ResponseEntity.ok("配置已刷新");
    }
}

// 配置变更监听器
@Component
public class ConfigChangeListener {
    
    private static final Logger logger = LoggerFactory.getLogger(ConfigChangeListener.class);
    
    @EventListener
    public void handleRefreshEvent(RefreshRemoteApplicationEvent event) {
        logger.info("配置刷新事件: {}", event.getDestinationService());
        
        // 处理配置变更后的逻辑
        // 例如：清除缓存、重新初始化组件等
    }
    
    @EventListener
    public void handleEnvironmentChangeEvent(EnvironmentChangeEvent event) {
        logger.info("环境变更事件，变更的键: {}", event.getKeys());
        
        // 处理特定配置项的变更
        for (String key : event.getKeys()) {
            if (key.startsWith("user-service.database")) {
                logger.info("数据库配置发生变更，需要重新初始化连接池");
                // 重新初始化数据库连接池
            }
        }
    }
}

分布式事务处理

Saga模式实现

// 订单Saga编排器
@Component
public class OrderSagaOrchestrator {
    
    private final UserServiceClient userServiceClient;
    private final InventoryServiceClient inventoryServiceClient;
    private final PaymentServiceClient paymentServiceClient;
    private final OrderRepository orderRepository;
    
    public OrderSagaOrchestrator(UserServiceClient userServiceClient,
                                InventoryServiceClient inventoryServiceClient,
                                PaymentServiceClient paymentServiceClient,
                                OrderRepository orderRepository) {
        this.userServiceClient = userServiceClient;
        this.inventoryServiceClient = inventoryServiceClient;
        this.paymentServiceClient = paymentServiceClient;
        this.orderRepository = orderRepository;
    }
    
    @Transactional
    public OrderResult processOrder(CreateOrderRequest request) {
        Order order = null;
        List<CompensationAction> compensations = new ArrayList<>();
        
        try {
            // 步骤1: 验证用户
            UserResponse user = userServiceClient.getUser(request.getUserId());
            if (user == null) {
                throw new UserNotFoundException("用户不存在: " + request.getUserId());
            }
            
            // 步骤2: 创建订单
            order = createOrder(request, user);
            compensations.add(() -> cancelOrder(order.getId()));
            
            // 步骤3: 预留库存
            ReservationResult reservation = inventoryServiceClient.reserveItems(
                new ReserveItemsRequest(order.getId(), request.getItems())
            );
            compensations.add(() -> inventoryServiceClient.releaseReservation(reservation.getReservationId()));
            
            // 步骤4: 处理支付
            PaymentResult payment = paymentServiceClient.processPayment(
                new ProcessPaymentRequest(order.getId(), order.getTotalAmount(), request.getPaymentMethod())
            );
            compensations.add(() -> paymentServiceClient.refundPayment(payment.getPaymentId()));
            
            // 步骤5: 确认库存
            inventoryServiceClient.confirmReservation(reservation.getReservationId());
            
            // 步骤6: 完成订单
            order.setStatus(OrderStatus.COMPLETED);
            order.setPaymentId(payment.getPaymentId());
            orderRepository.save(order);
            
            return new OrderResult(true, order, null);
            
        } catch (Exception e) {
            // 执行补偿操作
            executeCompensations(compensations);
            
            if (order != null) {
                order.setStatus(OrderStatus.FAILED);
                orderRepository.save(order);
            }
            
            return new OrderResult(false, order, e.getMessage());
        }
    }
    
    private Order createOrder(CreateOrderRequest request, UserResponse user) {
        Order order = new Order();
        order.setUserId(user.getId());
        order.setUserName(user.getUsername());
        order.setItems(request.getItems());
        order.setTotalAmount(calculateTotalAmount(request.getItems()));
        order.setStatus(OrderStatus.PENDING);
        
        return orderRepository.save(order);
    }
    
    private void cancelOrder(Long orderId) {
        try {
            Order order = orderRepository.findById(orderId).orElse(null);
            if (order != null) {
                order.setStatus(OrderStatus.CANCELLED);
                orderRepository.save(order);
            }
        } catch (Exception e) {
            // 记录补偿失败日志
            System.err.println("取消订单失败: " + orderId + ", 错误: " + e.getMessage());
        }
    }
    
    private void executeCompensations(List<CompensationAction> compensations) {
        // 逆序执行补偿操作
        Collections.reverse(compensations);
        
        for (CompensationAction compensation : compensations) {
            try {
                compensation.execute();
            } catch (Exception e) {
                // 记录补偿失败日志，但继续执行其他补偿
                System.err.println("补偿操作失败: " + e.getMessage());
            }
        }
    }
    
    private BigDecimal calculateTotalAmount(List<OrderItem> items) {
        return items.stream()
            .map(item -> item.getPrice().multiply(BigDecimal.valueOf(item.getQuantity())))
            .reduce(BigDecimal.ZERO, BigDecimal::add);
    }
    
    @FunctionalInterface
    private interface CompensationAction {
        void execute();
    }
}

// 库存服务客户端
@FeignClient(name = "inventory-service")
public interface InventoryServiceClient {
    
    @PostMapping("/api/inventory/reserve")
    ReservationResult reserveItems(@RequestBody ReserveItemsRequest request);
    
    @PostMapping("/api/inventory/confirm/{reservationId}")
    void confirmReservation(@PathVariable("reservationId") String reservationId);
    
    @PostMapping("/api/inventory/release/{reservationId}")
    void releaseReservation(@PathVariable("reservationId") String reservationId);
}

// 支付服务客户端
@FeignClient(name = "payment-service")
public interface PaymentServiceClient {
    
    @PostMapping("/api/payments/process")
    PaymentResult processPayment(@RequestBody ProcessPaymentRequest request);
    
    @PostMapping("/api/payments/refund/{paymentId}")
    void refundPayment(@PathVariable("paymentId") String paymentId);
}

// Saga状态管理
@Entity
@Table(name = "saga_transactions")
public class SagaTransaction {
    @Id
    private String sagaId;
    
    @Enumerated(EnumType.STRING)
    private SagaStatus status;
    
    @Column(columnDefinition = "TEXT")
    private String currentStep;
    
    @Column(columnDefinition = "TEXT")
    private String compensationData;
    
    @CreationTimestamp
    private LocalDateTime createdAt;
    
    @UpdateTimestamp
    private LocalDateTime updatedAt;
    
    // getters and setters
}

public enum SagaStatus {
    STARTED, IN_PROGRESS, COMPLETED, COMPENSATING, COMPENSATED, FAILED
}

// Saga事务管理器
@Service
public class SagaTransactionManager {
    
    private final SagaTransactionRepository sagaRepository;
    
    public SagaTransactionManager(SagaTransactionRepository sagaRepository) {
        this.sagaRepository = sagaRepository;
    }
    
    public SagaTransaction startSaga(String sagaType, String businessId) {
        SagaTransaction saga = new SagaTransaction();
        saga.setSagaId(generateSagaId(sagaType, businessId));
        saga.setStatus(SagaStatus.STARTED);
        saga.setCurrentStep("INIT");
        
        return sagaRepository.save(saga);
    }
    
    public void updateSagaStep(String sagaId, String step, SagaStatus status) {
        SagaTransaction saga = sagaRepository.findById(sagaId)
            .orElseThrow(() -> new SagaNotFoundException("Saga不存在: " + sagaId));
        
        saga.setCurrentStep(step);
        saga.setStatus(status);
        
        sagaRepository.save(saga);
    }
    
    public void completeSaga(String sagaId) {
        updateSagaStep(sagaId, "COMPLETED", SagaStatus.COMPLETED);
    }
    
    public void compensateSaga(String sagaId, String compensationData) {
        SagaTransaction saga = sagaRepository.findById(sagaId)
            .orElseThrow(() -> new SagaNotFoundException("Saga不存在: " + sagaId));
        
        saga.setStatus(SagaStatus.COMPENSATING);
        saga.setCompensationData(compensationData);
        
        sagaRepository.save(saga);
    }
    
    private String generateSagaId(String sagaType, String businessId) {
        return sagaType + "-" + businessId + "-" + System.currentTimeMillis();
    }
}

事件驱动架构

// 领域事件
public abstract class DomainEvent {
    private final String eventId;
    private final LocalDateTime occurredOn;
    private final String aggregateId;
    
    protected DomainEvent(String aggregateId) {
        this.eventId = UUID.randomUUID().toString();
        this.occurredOn = LocalDateTime.now();
        this.aggregateId = aggregateId;
    }
    
    // getters
    public String getEventId() { return eventId; }
    public LocalDateTime getOccurredOn() { return occurredOn; }
    public String getAggregateId() { return aggregateId; }
}

// 用户事件
public class UserCreatedEvent extends DomainEvent {
    private final Long userId;
    private final String username;
    private final String email;
    
    public UserCreatedEvent(Long userId, String username, String email) {
        super(userId.toString());
        this.userId = userId;
        this.username = username;
        this.email = email;
    }
    
    // getters
    public Long getUserId() { return userId; }
    public String getUsername() { return username; }
    public String getEmail() { return email; }
}

public class UserUpdatedEvent extends DomainEvent {
    private final Long userId;
    private final String oldEmail;
    private final String newEmail;
    
    public UserUpdatedEvent(Long userId, String oldEmail, String newEmail) {
        super(userId.toString());
        this.userId = userId;
        this.oldEmail = oldEmail;
        this.newEmail = newEmail;
    }
    
    // getters
    public Long getUserId() { return userId; }
    public String getOldEmail() { return oldEmail; }
    public String getNewEmail() { return newEmail; }
}

// 订单事件
public class OrderCreatedEvent extends DomainEvent {
    private final Long orderId;
    private final Long userId;
    private final BigDecimal totalAmount;
    private final List<OrderItem> items;
    
    public OrderCreatedEvent(Long orderId, Long userId, BigDecimal totalAmount, List<OrderItem> items) {
        super(orderId.toString());
        this.orderId = orderId;
        this.userId = userId;
        this.totalAmount = totalAmount;
        this.items = new ArrayList<>(items);
    }
    
    // getters
    public Long getOrderId() { return orderId; }
    public Long getUserId() { return userId; }
    public BigDecimal getTotalAmount() { return totalAmount; }
    public List<OrderItem> getItems() { return Collections.unmodifiableList(items); }
}

// 事件发布器
@Component
public class DomainEventPublisher {
    
    private final ApplicationEventPublisher applicationEventPublisher;
    private final MessageProducer messageProducer;
    
    public DomainEventPublisher(ApplicationEventPublisher applicationEventPublisher,
                               MessageProducer messageProducer) {
        this.applicationEventPublisher = applicationEventPublisher;
        this.messageProducer = messageProducer;
    }
    
    public void publish(DomainEvent event) {
        // 本地事件发布
        applicationEventPublisher.publishEvent(event);
        
        // 远程事件发布
        publishToMessageBroker(event);
    }
    
    private void publishToMessageBroker(DomainEvent event) {
        try {
            String topic = getTopicForEvent(event);
            String message = serializeEvent(event);
            
            messageProducer.send(topic, message);
        } catch (Exception e) {
            // 记录发布失败日志
            System.err.println("事件发布失败: " + event.getEventId() + ", 错误: " + e.getMessage());
        }
    }
    
    private String getTopicForEvent(DomainEvent event) {
        return event.getClass().getSimpleName().toLowerCase().replace("event", "");
    }
    
    private String serializeEvent(DomainEvent event) {
        // 使用JSON序列化事件
        ObjectMapper mapper = new ObjectMapper();
        try {
            return mapper.writeValueAsString(event);
        } catch (Exception e) {
            throw new EventSerializationException("事件序列化失败", e);
        }
    }
}

// 事件处理器
@Component
public class UserEventHandler {
    
    private final EmailService emailService;
    private final UserProfileService userProfileService;
    
    public UserEventHandler(EmailService emailService, UserProfileService userProfileService) {
        this.emailService = emailService;
        this.userProfileService = userProfileService;
    }
    
    @EventListener
    @Async
    public void handleUserCreated(UserCreatedEvent event) {
        try {
            // 发送欢迎邮件
            emailService.sendWelcomeEmail(event.getEmail(), event.getUsername());
            
            // 创建用户档案
            userProfileService.createProfile(event.getUserId(), event.getUsername());
            
            System.out.println("用户创建事件处理完成: " + event.getUserId());
        } catch (Exception e) {
            System.err.println("处理用户创建事件失败: " + e.getMessage());
        }
    }
    
    @EventListener
    @Async
    public void handleUserUpdated(UserUpdatedEvent event) {
        try {
            // 更新相关服务中的用户信息
            userProfileService.updateEmail(event.getUserId(), event.getNewEmail());
            
            System.out.println("用户更新事件处理完成: " + event.getUserId());
        } catch (Exception e) {
            System.err.println("处理用户更新事件失败: " + e.getMessage());
        }
    }
}

@Component
public class OrderEventHandler {
    
    private final InventoryService inventoryService;
    private final NotificationService notificationService;
    
    public OrderEventHandler(InventoryService inventoryService, 
                            NotificationService notificationService) {
        this.inventoryService = inventoryService;
        this.notificationService = notificationService;
    }
    
    @EventListener
    @Async
    public void handleOrderCreated(OrderCreatedEvent event) {
        try {
            // 更新库存
            inventoryService.updateInventory(event.getItems());
            
            // 发送订单确认通知
            notificationService.sendOrderConfirmation(event.getUserId(), event.getOrderId());
            
            System.out.println("订单创建事件处理完成: " + event.getOrderId());
        } catch (Exception e) {
            System.err.println("处理订单创建事件失败: " + e.getMessage());
        }
    }
}

// 消息生产者
@Component
public class MessageProducer {
    
    private final RabbitTemplate rabbitTemplate;
    
    public MessageProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }
    
    public void send(String topic, String message) {
        rabbitTemplate.convertAndSend(topic, message);
    }
}

// 消息消费者
@RabbitListener(queues = "user.created")
@Component
public class UserCreatedMessageConsumer {
    
    private final UserSyncService userSyncService;
    
    public UserCreatedMessageConsumer(UserSyncService userSyncService) {
        this.userSyncService = userSyncService;
    }
    
    @RabbitHandler
    public void handleUserCreatedMessage(String message) {
        try {
            ObjectMapper mapper = new ObjectMapper();
            UserCreatedEvent event = mapper.readValue(message, UserCreatedEvent.class);
            
            // 同步用户信息到其他服务
            userSyncService.syncUser(event.getUserId(), event.getUsername(), event.getEmail());
            
            System.out.println("用户创建消息处理完成: " + event.getUserId());
        } catch (Exception e) {
            System.err.println("处理用户创建消息失败: " + e.getMessage());
        }
    }
}

微服务监控与运维

健康检查

// 自定义健康检查
@Component
public class DatabaseHealthIndicator implements HealthIndicator {
    
    private final DataSource dataSource;
    
    public DatabaseHealthIndicator(DataSource dataSource) {
        this.dataSource = dataSource;
    }
    
    @Override
    public Health health() {
        try (Connection connection = dataSource.getConnection()) {
            if (connection.isValid(1)) {
                return Health.up()
                    .withDetail("database", "Available")
                    .withDetail("validationQuery", "SELECT 1")
                    .build();
            } else {
                return Health.down()
                    .withDetail("database", "Connection invalid")
                    .build();
            }
        } catch (Exception e) {
            return Health.down()
                .withDetail("database", "Connection failed")
                .withException(e)
                .build();
        }
    }
}

// 外部服务健康检查
@Component
public class ExternalServiceHealthIndicator implements HealthIndicator {
    
    private final UserServiceClient userServiceClient;
    
    public ExternalServiceHealthIndicator(UserServiceClient userServiceClient) {
        this.userServiceClient = userServiceClient;
    }
    
    @Override
    public Health health() {
        try {
            // 调用外部服务的健康检查端点
            ResponseEntity<String> response = userServiceClient.healthCheck();
            
            if (response.getStatusCode().is2xxSuccessful()) {
                return Health.up()
                    .withDetail("userService", "Available")
                    .withDetail("responseTime", measureResponseTime())
                    .build();
            } else {
                return Health.down()
                    .withDetail("userService", "Unhealthy")
                    .withDetail("statusCode", response.getStatusCode())
                    .build();
            }
        } catch (Exception e) {
            return Health.down()
                .withDetail("userService", "Unavailable")
                .withException(e)
                .build();
        }
    }
    
    private long measureResponseTime() {
        long startTime = System.currentTimeMillis();
        try {
            userServiceClient.healthCheck();
            return System.currentTimeMillis() - startTime;
        } catch (Exception e) {
            return -1;
        }
    }
}

// 聚合健康检查
@RestController
@RequestMapping("/actuator/health")
public class HealthController {
    
    private final HealthEndpoint healthEndpoint;
    
    public HealthController(HealthEndpoint healthEndpoint) {
        this.healthEndpoint = healthEndpoint;
    }
    
    @GetMapping
    public ResponseEntity<Health> health() {
        Health health = healthEndpoint.health();
        HttpStatus status = health.getStatus() == Status.UP ? HttpStatus.OK : HttpStatus.SERVICE_UNAVAILABLE;
        return ResponseEntity.status(status).body(health);
    }
    
    @GetMapping("/detailed")
    public ResponseEntity<Map<String, Object>> detailedHealth() {
        Health health = healthEndpoint.health();
        Map<String, Object> details = new HashMap<>();
        details.put("status", health.getStatus());
        details.put("components", health.getDetails());
        details.put("timestamp", System.currentTimeMillis());
        
        HttpStatus status = health.getStatus() == Status.UP ? HttpStatus.OK : HttpStatus.SERVICE_UNAVAILABLE;
        return ResponseEntity.status(status).body(details);
    }
}

链路追踪

// Sleuth配置
@Configuration
public class TracingConfiguration {
    
    @Bean
    public Sampler alwaysSampler() {
        return Sampler.create(1.0f); // 100%采样
    }
    
    @Bean
    public SpanCustomizer spanCustomizer() {
        return span -> {
            span.tag("service.name", "user-service");
            span.tag("service.version", "1.0.0");
        };
    }
}

// 自定义追踪
@Service
public class TracedUserService {
    
    private final UserRepository userRepository;
    private final Tracer tracer;
    
    public TracedUserService(UserRepository userRepository, Tracer tracer) {
        this.userRepository = userRepository;
        this.tracer = tracer;
    }
    
    @NewSpan("user-creation")
    public User createUser(@SpanTag("username") String username, 
                          @SpanTag("email") String email) {
        
        Span span = tracer.nextSpan()
            .name("validate-user-data")
            .tag("operation", "validation")
            .start();
        
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            // 验证用户数据
            validateUserData(username, email);
            span.tag("validation.result", "success");
        } catch (Exception e) {
            span.tag("validation.result", "failed");
            span.tag("error", e.getMessage());
            throw e;
        } finally {
            span.end();
        }
        
        // 创建用户
        return createUserInDatabase(username, email);
    }
    
    @NewSpan("database-operation")
    private User createUserInDatabase(@SpanTag("username") String username, 
                                     @SpanTag("email") String email) {
        User user = new User(username, email, "defaultPassword");
        return userRepository.save(user);
    }
    
    private void validateUserData(String username, String email) {
        if (username == null || username.trim().isEmpty()) {
            throw new IllegalArgumentException("用户名不能为空");
        }
        if (email == null || !email.contains("@")) {
            throw new IllegalArgumentException("邮箱格式不正确");
        }
    }
}

// 手动创建Span
@Service
public class ManualTracingService {
    
    private final Tracer tracer;
    private final UserServiceClient userServiceClient;
    
    public ManualTracingService(Tracer tracer, UserServiceClient userServiceClient) {
        this.tracer = tracer;
        this.userServiceClient = userServiceClient;
    }
    
    public void processUserOperation(Long userId) {
        Span span = tracer.nextSpan()
            .name("process-user-operation")
            .tag("user.id", userId.toString())
            .tag("operation.type", "user-processing")
            .start();
        
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            
            // 子操作1：获取用户信息
            Span getUserSpan = tracer.nextSpan()
                .name("get-user-info")
                .tag("user.id", userId.toString())
                .start();
            
            UserResponse user;
            try (Tracer.SpanInScope getUserWs = tracer.withSpanInScope(getUserSpan)) {
                user = userServiceClient.getUser(userId);
                getUserSpan.tag("user.found", "true");
                getUserSpan.tag("user.name", user.getUsername());
            } catch (Exception e) {
                getUserSpan.tag("user.found", "false");
                getUserSpan.tag("error", e.getMessage());
                throw e;
            } finally {
                getUserSpan.end();
            }
            
            // 子操作2：处理业务逻辑
            Span processSpan = tracer.nextSpan()
                .name("business-logic-processing")
                .tag("user.id", userId.toString())
                .start();
            
            try (Tracer.SpanInScope processWs = tracer.withSpanInScope(processSpan)) {
                // 模拟业务处理
                Thread.sleep(100);
                processSpan.tag("processing.result", "success");
            } catch (Exception e) {
                processSpan.tag("processing.result", "failed");
                processSpan.tag("error", e.getMessage());
                throw e;
            } finally {
                processSpan.end();
            }
            
            span.tag("operation.result", "completed");
            
        } catch (Exception e) {
            span.tag("operation.result", "failed");
            span.tag("error", e.getMessage());
            throw new RuntimeException("用户操作处理失败", e);
        } finally {
            span.end();
        }
    }
}

性能监控

// Micrometer指标配置
@Configuration
public class MetricsConfiguration {
    
    @Bean
    public TimedAspect timedAspect(MeterRegistry registry) {
        return new TimedAspect(registry);
    }
    
    @Bean
    public CountedAspect countedAspect(MeterRegistry registry) {
        return new CountedAspect(registry);
    }
    
    // 自定义指标
    @Bean
    public Counter userCreationCounter(MeterRegistry registry) {
        return Counter.builder("user.creation.total")
            .description("用户创建总数")
            .register(registry);
    }
    
    @Bean
    public Timer userCreationTimer(MeterRegistry registry) {
        return Timer.builder("user.creation.duration")
            .description("用户创建耗时")
            .register(registry);
    }
    
    @Bean
    public Gauge activeUsersGauge(MeterRegistry registry, UserService userService) {
        return Gauge.builder("user.active.count")
            .description("活跃用户数量")
            .register(registry, userService, UserService::getActiveUserCount);
    }
}

// 使用指标的服务
@Service
public class MetricUserService {
    
    private final UserRepository userRepository;
    private final Counter userCreationCounter;
    private final Timer userCreationTimer;
    private final MeterRegistry meterRegistry;
    
    public MetricUserService(UserRepository userRepository,
                            Counter userCreationCounter,
                            Timer userCreationTimer,
                            MeterRegistry meterRegistry) {
        this.userRepository = userRepository;
        this.userCreationCounter = userCreationCounter;
        this.userCreationTimer = userCreationTimer;
        this.meterRegistry = meterRegistry;
    }
    
    @Timed(name = "user.creation", description = "用户创建耗时")
    @Counted(name = "user.creation.attempts", description = "用户创建尝试次数")
    public User createUser(CreateUserRequest request) {
        return Timer.Sample.start(meterRegistry)
            .stop(userCreationTimer.wrap(() -> {
                try {
                    User user = new User(request.getUsername(), request.getEmail(), request.getPassword());
                    User savedUser = userRepository.save(user);
                    
                    // 增加成功计数
                    userCreationCounter.increment(
                        Tags.of(
                            Tag.of("result", "success"),
                            Tag.of("source", "api")
                        )
                    );
                    
                    return savedUser;
                } catch (Exception e) {
                    // 增加失败计数
                    userCreationCounter.increment(
                        Tags.of(
                            Tag.of("result", "failure"),
                            Tag.of("error", e.getClass().getSimpleName())
                        )
                    );
                    throw e;
                }
            }));
    }
    
    public long getActiveUserCount() {
        // 计算活跃用户数量的逻辑
        return userRepository.countByLastLoginAfter(
            LocalDateTime.now().minusDays(7)
        );
    }
    
    // 自定义业务指标
    public void recordUserLogin(Long userId, String source) {
        meterRegistry.counter("user.login.total",
            Tags.of(
                Tag.of("source", source),
                Tag.of("user.type", getUserType(userId))
            )
        ).increment();
    }
    
    public void recordApiCall(String endpoint, String method, int statusCode, long duration) {
        Timer.builder("api.request.duration")
            .tag("endpoint", endpoint)
            .tag("method", method)
            .tag("status", String.valueOf(statusCode))
            .register(meterRegistry)
            .record(duration, TimeUnit.MILLISECONDS);
    }
    
    private String getUserType(Long userId) {
        // 根据用户ID判断用户类型
        return userId < 1000 ? "premium" : "regular";
    }
}

// 自定义指标端点
@RestController
@RequestMapping("/metrics")
public class CustomMetricsController {
    
    private final MeterRegistry meterRegistry;
    private final UserService userService;
    
    public CustomMetricsController(MeterRegistry meterRegistry, UserService userService) {
        this.meterRegistry = meterRegistry;
        this.userService = userService;
    }
    
    @GetMapping("/business")
    public ResponseEntity<Map<String, Object>> getBusinessMetrics() {
        Map<String, Object> metrics = new HashMap<>();
        
        // 用户相关指标
        metrics.put("totalUsers", getTotalUsers());
        metrics.put("activeUsers", getActiveUsers());
        metrics.put("newUsersToday", getNewUsersToday());
        
        // 系统性能指标
        metrics.put("memoryUsage", getMemoryUsage());
        metrics.put("cpuUsage", getCpuUsage());
        
        // 业务指标
        metrics.put("apiCallsPerMinute", getApiCallsPerMinute());
        metrics.put("errorRate", getErrorRate());
        
        return ResponseEntity.ok(metrics);
    }
    
    private long getTotalUsers() {
        return meterRegistry.get("user.creation.total").counter().count();
    }
    
    private long getActiveUsers() {
        return (long) meterRegistry.get("user.active.count").gauge().value();
    }
    
    private long getNewUsersToday() {
        // 计算今日新增用户数
        return userService.countUsersCreatedAfter(LocalDate.now().atStartOfDay());
    }
    
    private double getMemoryUsage() {
        return meterRegistry.get("jvm.memory.used").gauge().value();
    }
    
    private double getCpuUsage() {
        return meterRegistry.get("system.cpu.usage").gauge().value();
    }
    
    private double getApiCallsPerMinute() {
        return meterRegistry.get("api.request.duration").timer().count() / 60.0;
    }
    
    private double getErrorRate() {
        Timer timer = meterRegistry.get("api.request.duration").timer();
        long totalRequests = (long) timer.count();
        long errorRequests = meterRegistry.get("api.request.duration")
            .tag("status", "500")
            .timer()
            .count();
        
        return totalRequests > 0 ? (double) errorRequests / totalRequests : 0.0;
    }
}

微服务部署与DevOps

Docker容器化

# Dockerfile
FROM openjdk:11-jre-slim

# 设置工作目录
WORKDIR /app

# 复制jar文件
COPY target/user-service-1.0.0.jar app.jar

# 创建非root用户
RUN addgroup --system spring && adduser --system spring --ingroup spring
USER spring:spring

# 暴露端口
EXPOSE 8080

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8080/actuator/health || exit 1

# 启动应用
ENTRYPOINT ["java", "-jar", "/app/app.jar"]

# docker-compose.yml
version: '3.8'

services:
  # 服务注册中心
  eureka-server:
    image: user-service/eureka-server:latest
    ports:
      - "8761:8761"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
    networks:
      - microservices-network

  # 配置中心
  config-server:
    image: user-service/config-server:latest
    ports:
      - "8888:8888"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
      - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka
    depends_on:
      - eureka-server
    networks:
      - microservices-network

  # API网关
  api-gateway:
    image: user-service/api-gateway:latest
    ports:
      - "8080:8080"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
      - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka
    depends_on:
      - eureka-server
      - config-server
    networks:
      - microservices-network

  # 用户服务
  user-service:
    image: user-service/user-service:latest
    ports:
      - "8081:8081"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
      - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka
      - SPRING_CLOUD_CONFIG_URI=http://config-server:8888
      - SPRING_DATASOURCE_URL=jdbc:mysql://mysql:3306/user_service
      - SPRING_DATASOURCE_USERNAME=root
      - SPRING_DATASOURCE_PASSWORD=password
    depends_on:
      - eureka-server
      - config-server
      - mysql
    networks:
      - microservices-network

  # 订单服务
  order-service:
    image: user-service/order-service:latest
    ports:
      - "8082:8082"
    environment:
      - SPRING_PROFILES_ACTIVE=docker
      - EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE=http://eureka-server:8761/eureka
      - SPRING_CLOUD_CONFIG_URI=http://config-server:8888
      - SPRING_DATASOURCE_URL=jdbc:mysql://mysql:3306/order_service
    depends_on:
      - eureka-server
      - config-server
      - mysql
      - user-service
    networks:
      - microservices-network

  # MySQL数据库
  mysql:
    image: mysql:8.0
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=password
      - MYSQL_DATABASE=user_service
    volumes:
      - mysql-data:/var/lib/mysql
      - ./init-scripts:/docker-entrypoint-initdb.d
    networks:
      - microservices-network

  # Redis缓存
  redis:
    image: redis:6-alpine
    ports:
      - "6379:6379"
    networks:
      - microservices-network

  # RabbitMQ消息队列
  rabbitmq:
    image: rabbitmq:3-management
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      - RABBITMQ_DEFAULT_USER=admin
      - RABBITMQ_DEFAULT_PASS=password
    networks:
      - microservices-network

volumes:
  mysql-data:

networks:
  microservices-network:
    driver: bridge

Kubernetes部署

# user-service-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: user-service
  labels:
    app: user-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: user-service
  template:
    metadata:
      labels:
        app: user-service
    spec:
      containers:
      - name: user-service
        image: user-service/user-service:1.0.0
        ports:
        - containerPort: 8081
        env:
        - name: SPRING_PROFILES_ACTIVE
          value: "kubernetes"
        - name: EUREKA_CLIENT_SERVICE_URL_DEFAULTZONE
          value: "http://eureka-server:8761/eureka"
        - name: SPRING_DATASOURCE_URL
          valueFrom:
            secretKeyRef:
              name: database-secret
              key: url
        - name: SPRING_DATASOURCE_USERNAME
          valueFrom:
            secretKeyRef:
              name: database-secret
              key: username
        - name: SPRING_DATASOURCE_PASSWORD
          valueFrom:
            secretKeyRef:
              name: database-secret
              key: password
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "1Gi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /actuator/health
            port: 8081
          initialDelaySeconds: 60
          periodSeconds: 30
        readinessProbe:
          httpGet:
            path: /actuator/health
            port: 8081
          initialDelaySeconds: 30
          periodSeconds: 10

---
apiVersion: v1
kind: Service
metadata:
  name: user-service
spec:
  selector:
    app: user-service
  ports:
  - protocol: TCP
    port: 8081
    targetPort: 8081
  type: ClusterIP

---
apiVersion: v1
kind: Secret
metadata:
  name: database-secret
type: Opaque
data:
  url: amRiYzpteXNxbDovL215c3FsOjMzMDYvdXNlcl9zZXJ2aWNl  # base64编码
  username: cm9vdA==  # base64编码的root
  password: cGFzc3dvcmQ=  # base64编码的password

总结

Java微服务架构通过将大型应用拆分为多个小型、独立的服务，实现了更好的可扩展性、可维护性和技术多样性。本文详细介绍了：

核心技术栈

• Spring Boot：快速构建微服务应用
• Spring Cloud：提供微服务治理能力
• Eureka：服务注册与发现
• Feign：声明式服务调用
• Hystrix：断路器模式实现
• Gateway：API网关统一入口

关键设计原则

1. 单一职责：每个服务专注于特定业务功能
2. 自治性：服务独立开发、部署和扩展
3. 去中心化：避免单点故障
4. 容错性：优雅处理服务故障
5. 可观测性：全面的监控和追踪

最佳实践

• 合理的服务拆分粒度
• 完善的错误处理和降级机制
• 统一的配置管理
• 全链路监控和追踪
• 自动化部署和运维

微服务架构虽然带来了分布式系统的复杂性，但通过合理的设计和工具选择，可以构建出高可用、高性能的现代化应用系统。在实际项目中，需要根据业务需求和团队能力，逐步演进和优化微服务架构。