前端技术积累

外观

阶段五：微服务架构（4-6周） ​

作为资深前端开发者，你已经熟悉模块化开发和组件化思想。本阶段将学习如何将单体应用拆分为微服务架构，掌握Spring Cloud生态、消息队列和分布式系统基础。

5.1 Spring Cloud生态（第1-21天） ​

5.1.1 微服务架构概述 ​

什么是微服务 ​

微服务架构是一种将单体应用程序拆分为一组小型服务的方法，每个服务运行在自己的进程中，通过轻量级机制（通常是HTTP REST API）进行通信。

与前端对比：

前端概念微服务概念
单页应用（SPA）单体应用
微前端（qiankun/module federation）微服务
组件拆分服务拆分
npm包管理Maven依赖管理
API网关（Nginx）Spring Cloud Gateway
服务注册（Consul）Nacos/Eureka
状态管理（Redux/Vuex）分布式配置中心

微服务架构图 ​

┌─────────────────────────────────────────────────────────────┐
│                        客户端（Web/App）                      │
└────────────────────────┬────────────────────────────────────┘
                         │
┌────────────────────────▼────────────────────────────────────┐
│                   API网关（Spring Cloud Gateway）             │
│              - 路由转发、鉴权、限流、日志                       │
└────────┬──────────────┬──────────────┬──────────────────────┘
         │              │              │
    ┌────▼────┐    ┌────▼────┐    ┌────▼────┐
    │ 用户服务 │    │ 订单服务 │    │ 商品服务 │  <- 各个微服务
    │ (8081)  │    │ (8082)  │    │ (8083)  │
    └────┬────┘    └────┬────┘    └────┬────┘
         │              │              │
    ┌────▼──────────────▼──────────────▼────┐
    │        服务注册与发现（Nacos）           │
    │     - 服务注册、服务发现、健康检查        │
    └─────────────────┬─────────────────────┘
                      │
         ┌────────────┼────────────┐
         │            │            │
    ┌────▼────┐  ┌────▼────┐  ┌───▼─────┐
    │  MySQL  │  │  Redis  │  │ RabbitMQ│  <- 基础设施
    │ (数据库) │  │ (缓存)  │  │ (消息)  │
    └─────────┘  └─────────┘  └─────────┘

5.1.2 Spring Cloud Alibaba简介 ​

Spring Cloud Alibaba是阿里巴巴提供的微服务解决方案，是国内主流的微服务框架套件。

核心组件 ​

组件功能对标Netflix
Nacos服务注册与发现、配置中心Eureka + Config
OpenFeign服务调用Feign
Sentinel熔断降级、流量控制Hystrix
Seata分布式事务无
Spring Cloud GatewayAPI网关Zuul
Spring Cloud LoadBalancer负载均衡Ribbon

Maven父工程配置 ​

xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>microservice-parent</artifactId>
    <version>1.0.0</version>
    <packaging>pom</packaging>
    
    <name>微服务父工程</name>
    <description>微服务架构父工程，统一管理依赖版本</description>

    <!-- 子模块 -->
    <modules>
        <module>common</module>
        <module>gateway</module>
        <module>service-user</module>
        <module>service-order</module>
        <module>service-product</module>
    </modules>

    <!-- 版本管理 -->
    <properties>
        <java.version>17</java.version>
        <maven.compiler.source>17</maven.compiler.source>
        <maven.compiler.target>17</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        
        <!-- Spring Boot版本 -->
        <spring-boot.version>3.2.0</spring-boot.version>
        
        <!-- Spring Cloud版本 -->
        <spring-cloud.version>2023.0.0</spring-cloud.version>
        
        <!-- Spring Cloud Alibaba版本 -->
        <spring-cloud-alibaba.version>2023.0.0.0</spring-cloud-alibaba.version>
    </properties>

    <!-- 依赖版本管理 -->
    <dependencyManagement>
        <dependencies>
            <!-- Spring Boot -->
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            
            <!-- Spring Cloud -->
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            
            <!-- Spring Cloud Alibaba -->
            <dependency>
                <groupId>com.alibaba.cloud</groupId>
                <artifactId>spring-cloud-alibaba-dependencies</artifactId>
                <version>${spring-cloud-alibaba.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

5.1.3 Nacos服务注册与发现 ​

Nacos简介 ​

Nacos是阿里巴巴开源的服务注册中心和配置中心，类似于前端的Consul或Etcd。

与前端对比：

前端Nacos
环境变量配置配置中心
服务发现（DNS）服务注册与发现
健康检查服务健康检查

安装Nacos ​

  1. 下载Nacos：Nacos官网
  2. 解压到指定目录
  3. 启动（单机模式）：
bash
# Windows
startup.cmd -m standalone

# Linux/Mac
sh startup.sh -m standalone
  1. 访问控制台：http://localhost:8848/nacos
    • 默认账号密码：nacos/nacos

服务提供者配置 ​

Maven依赖（service-user模块）：

xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>com.example</groupId>
        <artifactId>microservice-parent</artifactId>
        <version>1.0.0</version>
    </parent>

    <artifactId>service-user</artifactId>
    <packaging>jar</packaging>
    <name>用户服务</name>

    <dependencies>
        <!-- Spring Boot Web -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!-- Nacos服务发现 -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>

        <!-- Nacos配置中心 -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
        </dependency>

        <!-- Spring Cloud Bootstrap（配置加载） -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-bootstrap</artifactId>
        </dependency>

        <!-- Actuator健康检查 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>${spring-boot.version}</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

bootstrap.yml配置：

yaml
spring:
  application:
    name: service-user  # 服务名称（必须唯一）
  cloud:
    nacos:
      # 服务发现配置
      discovery:
        server-addr: localhost:8848  # Nacos服务器地址
        namespace: public  # 命名空间
        group: DEFAULT_GROUP  # 分组
      # 配置中心配置
      config:
        server-addr: localhost:8848
        namespace: public
        group: DEFAULT_GROUP
        file-extension: yaml  # 配置文件格式
        # 共享配置
        shared-configs:
          - data-id: common.yaml
            group: DEFAULT_GROUP
            refresh: true

application.yml配置：

yaml
server:
  port: 8081  # 服务端口

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/user_db?useSSL=false&serverTimezone=UTC&characterEncoding=utf8
    username: root
    password: 123456
    driver-class-name: com.mysql.cj.jdbc.Driver

# Actuator端点配置
management:
  endpoints:
    web:
      exposure:
        include: "*"  # 暴露所有端点
  endpoint:
    health:
      show-details: always

启动类：

java
package com.example.user;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;

@SpringBootApplication
@EnableDiscoveryClient  // 启用服务发现
public class UserServiceApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(UserServiceApplication.class, args);
        System.out.println("用户服务启动成功！");
    }
}

实体类：

java
package com.example.user.entity;

import lombok.Data;
import java.util.Date;

@Data
public class User {
    private Long id;
    private String username;
    private String email;
    private Integer age;
    private Date createdAt;
    private Date updatedAt;
}

Controller：

java
package com.example.user.controller;

import com.example.user.entity.User;
import org.springframework.web.bind.annotation.*;
import java.util.*;

@RestController
@RequestMapping("/api/users")
public class UserController {
    
    // 模拟数据库
    private static final Map<Long, User> USER_DB = new HashMap<>();
    
    static {
        User user1 = new User();
        user1.setId(1L);
        user1.setUsername("张三");
        user1.setEmail("zhangsan@example.com");
        user1.setAge(25);
        user1.setCreatedAt(new Date());
        USER_DB.put(1L, user1);
    }
    
    /**
     * 根据ID查询用户
     */
    @GetMapping("/{id}")
    public User getUserById(@PathVariable Long id) {
        return USER_DB.get(id);
    }
    
    /**
     * 查询所有用户
     */
    @GetMapping
    public List<User> getAllUsers() {
        return new ArrayList<>(USER_DB.values());
    }
    
    /**
     * 创建用户
     */
    @PostMapping
    public User createUser(@RequestBody User user) {
        Long id = USER_DB.size() + 1L;
        user.setId(id);
        user.setCreatedAt(new Date());
        USER_DB.put(id, user);
        return user;
    }
    
    /**
     * 更新用户
     */
    @PutMapping("/{id}")
    public User updateUser(@PathVariable Long id, @RequestBody User user) {
        user.setId(id);
        USER_DB.put(id, user);
        return user;
    }
    
    /**
     * 删除用户
     */
    @DeleteMapping("/{id}")
    public void deleteUser(@PathVariable Long id) {
        USER_DB.remove(id);
    }
}

服务消费者配置 ​

Maven依赖（service-order模块）：

xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>com.example</groupId>
        <artifactId>microservice-parent</artifactId>
        <version>1.0.0</version>
    </parent>

    <artifactId>service-order</artifactId>
    <packaging>jar</packaging>
    <name>订单服务</name>

    <dependencies>
        <!-- Spring Boot Web -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!-- Nacos服务发现 -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>

        <!-- OpenFeign服务调用 -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-openfeign</artifactId>
        </dependency>

        <!-- Spring Cloud LoadBalancer -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-loadbalancer</artifactId>
        </dependency>

        <!-- Actuator -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>

        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>${spring-boot.version}</version>
            </plugin>
        </plugins>
    </build>
</project>

bootstrap.yml：

yaml
spring:
  application:
    name: service-order
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848

application.yml：

yaml
server:
  port: 8082

# Feign配置
spring:
  cloud:
    openfeign:
      client:
        config:
          default:
            connect-timeout: 5000  # 连接超时时间（毫秒）
            read-timeout: 10000    # 读取超时时间（毫秒）
            logger-level: FULL     # 日志级别

启动类：

java
package com.example.order;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;

@SpringBootApplication
@EnableDiscoveryClient
@EnableFeignClients  // 启用Feign客户端
public class OrderServiceApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(OrderServiceApplication.class, args);
        System.out.println("订单服务启动成功！");
    }
}

5.1.4 OpenFeign服务调用 ​

OpenFeign简介 ​

OpenFeign是Spring Cloud提供的声明式REST客户端，简化了微服务之间的调用。类似于前端的axios封装。

与前端对比：

前端OpenFeign
axios/fetchOpenFeign
API封装FeignClient接口
拦截器Feign拦截器
请求/响应转换器Encoder/Decoder

定义Feign客户端 ​

创建Feign接口：

java
package com.example.order.feign;

import com.example.order.entity.User;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.*;
import java.util.List;

/**
 * 用户服务Feign客户端
 * name: 服务提供者的服务名称
 * fallback: 降级处理类
 */
@FeignClient(name = "service-user", fallback = UserFeignFallback.class)
public interface UserFeignClient {
    
    /**
     * 根据ID查询用户
     * 对应服务提供者的接口：GET /api/users/{id}
     */
    @GetMapping("/api/users/{id}")
    User getUserById(@PathVariable("id") Long id);
    
    /**
     * 查询所有用户
     */
    @GetMapping("/api/users")
    List<User> getAllUsers();
    
    /**
     * 创建用户
     */
    @PostMapping("/api/users")
    User createUser(@RequestBody User user);
    
    /**
     * 更新用户
     */
    @PutMapping("/api/users/{id}")
    User updateUser(@PathVariable("id") Long id, @RequestBody User user);
    
    /**
     * 删除用户
     */
    @DeleteMapping("/api/users/{id}")
    void deleteUser(@PathVariable("id") Long id);
}

Feign降级处理：

java
package com.example.order.feign;

import com.example.order.entity.User;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;

/**
 * 用户服务Feign降级处理
 * 当服务调用失败时，返回默认值
 */
@Component
public class UserFeignFallback implements UserFeignClient {
    
    @Override
    public User getUserById(Long id) {
        User fallbackUser = new User();
        fallbackUser.setId(id);
        fallbackUser.setUsername("默认用户");
        fallbackUser.setEmail("fallback@example.com");
        return fallbackUser;
    }
    
    @Override
    public List<User> getAllUsers() {
        return new ArrayList<>();
    }
    
    @Override
    public User createUser(User user) {
        return null;
    }
    
    @Override
    public User updateUser(Long id, User user) {
        return null;
    }
    
    @Override
    public void deleteUser(Long id) {
        // 降级处理
    }
}

Feign拦截器：

java
package com.example.order.feign;

import feign.RequestInterceptor;
import feign.RequestTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Feign配置类
 */
@Configuration
public class FeignConfig {
    
    /**
     * Feign拦截器：添加请求头
     */
    @Bean
    public RequestInterceptor requestInterceptor() {
        return new RequestInterceptor() {
            @Override
            public void apply(RequestTemplate template) {
                // 添加请求头
                template.header("X-Service-Source", "order-service");
                template.header("X-Request-Time", String.valueOf(System.currentTimeMillis()));
                
                // 可以添加Token等认证信息
                // template.header("Authorization", "Bearer " + token);
            }
        };
    }
}

使用Feign客户端 ​

订单实体类：

java
package com.example.order.entity;

import lombok.Data;
import java.math.BigDecimal;
import java.util.Date;

@Data
public class Order {
    private Long id;
    private String orderNo;
    private Long userId;
    private BigDecimal amount;
    private Integer status;  // 0-待支付，1-已支付，2-已取消
    private Date createdAt;
    
    // 关联用户信息（非数据库字段）
    private User user;
}

订单服务：

java
package com.example.order.service;

import com.example.order.entity.Order;
import com.example.order.entity.User;
import com.example.order.feign.UserFeignClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.util.*;

@Service
public class OrderService {
    
    @Autowired
    private UserFeignClient userFeignClient;
    
    // 模拟数据库
    private static final Map<Long, Order> ORDER_DB = new HashMap<>();
    
    static {
        Order order1 = new Order();
        order1.setId(1L);
        order1.setOrderNo("ORD20240101001");
        order1.setUserId(1L);
        order1.setAmount(new BigDecimal("99.99"));
        order1.setStatus(1);
        order1.setCreatedAt(new Date());
        ORDER_DB.put(1L, order1);
    }
    
    /**
     * 根据ID查询订单（包含用户信息）
     */
    public Order getOrderById(Long id) {
        Order order = ORDER_DB.get(id);
        if (order != null) {
            // 通过Feign调用用户服务获取用户信息
            User user = userFeignClient.getUserById(order.getUserId());
            order.setUser(user);
        }
        return order;
    }
    
    /**
     * 查询所有订单（包含用户信息）
     */
    public List<Order> getAllOrders() {
        List<Order> orders = new ArrayList<>(ORDER_DB.values());
        
        // 批量获取用户信息
        for (Order order : orders) {
            User user = userFeignClient.getUserById(order.getUserId());
            order.setUser(user);
        }
        
        return orders;
    }
    
    /**
     * 创建订单
     */
    public Order createOrder(Order order) {
        // 验证用户是否存在
        User user = userFeignClient.getUserById(order.getUserId());
        if (user == null) {
            throw new RuntimeException("用户不存在");
        }
        
        Long id = ORDER_DB.size() + 1L;
        order.setId(id);
        order.setOrderNo("ORD" + System.currentTimeMillis());
        order.setStatus(0);
        order.setCreatedAt(new Date());
        ORDER_DB.put(id, order);
        
        return order;
    }
}

订单控制器：

java
package com.example.order.controller;

import com.example.order.entity.Order;
import com.example.order.service.OrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.List;

@RestController
@RequestMapping("/api/orders")
public class OrderController {
    
    @Autowired
    private OrderService orderService;
    
    @GetMapping("/{id}")
    public Order getOrderById(@PathVariable Long id) {
        return orderService.getOrderById(id);
    }
    
    @GetMapping
    public List<Order> getAllOrders() {
        return orderService.getAllOrders();
    }
    
    @PostMapping
    public Order createOrder(@RequestBody Order order) {
        return orderService.createOrder(order);
    }
}

5.1.5 Spring Cloud LoadBalancer ​

负载均衡简介 ​

负载均衡是将请求分发到多个服务实例的机制。类似于前端CDN的负载均衡策略。

与前端对比：

前端LoadBalancer
Nginx负载均衡Spring Cloud LoadBalancer
轮询策略Round Robin
权重分配Weighted
随机选择Random

负载均衡策略配置 ​

application.yml配置：

yaml
spring:
  cloud:
    loadbalancer:
      # 负载均衡策略
      configurations: random  # 随机策略（默认是轮询）
      
      # 健康检查
      health-check:
        enabled: true
        initial-delay: 3s
        interval: 30s
        timeout: 5s

自定义负载均衡策略：

java
package com.example.order.config;

import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.loadbalancer.core.ReactorLoadBalancer;
import org.springframework.cloud.loadbalancer.core.RoundRobinLoadBalancer;
import org.springframework.cloud.loadbalancer.support.LoadBalancerClientFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

/**
 * 自定义负载均衡配置
 */
@Configuration
public class LoadBalancerConfig {
    
    /**
     * 自定义负载均衡器（轮询策略）
     */
    @Bean
    public ReactorLoadBalancer<ServiceInstance> reactorServiceInstanceLoadBalancer(
            Environment environment,
            LoadBalancerClientFactory loadBalancerClientFactory) {
        String name = environment.getProperty(LoadBalancerClientFactory.PROPERTY_NAME);
        return new RoundRobinLoadBalancer(
            loadBalancerClientFactory.getLazyProvider(name, ServiceInstanceListSupplier.class),
            name
        );
    }
}

5.1.6 Sentinel熔断降级 ​

Sentinel简介 ​

Sentinel是阿里巴巴开源的流量控制组件，提供流量控制、熔断降级、系统负载保护等功能。类似于前端的错误边界和降级策略。

与前端对比：

前端Sentinel
错误边界熔断降级
请求限流流量控制
降级UI服务降级
超时控制超时熔断

安装Sentinel ​

  1. 下载Sentinel Dashboard：Sentinel官网
  2. 启动Dashboard：
bash
java -jar sentinel-dashboard-1.8.6.jar
  1. 访问控制台：http://localhost:8080
    • 默认账号密码：sentinel/sentinel

Maven依赖 ​

xml
<!-- Sentinel依赖 -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
</dependency>

application.yml配置 ​

yaml
spring:
  cloud:
    sentinel:
      transport:
        dashboard: localhost:8080  # Sentinel Dashboard地址
        port: 8719  # 与Dashboard通信的端口
      # 启动时立即初始化
      eager: true

流量控制 ​

Controller示例：

java
package com.example.order.controller;

import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/api/orders")
public class OrderController {
    
    /**
     * 查询订单（带限流）
     * @SentinelResource 定义资源
     * blockHandler 限流/降级处理方法
     */
    @GetMapping("/test")
    @SentinelResource(value = "getOrderTest", blockHandler = "handleBlock")
    public String getOrderTest() {
        return "订单查询成功！";
    }
    
    /**
     * 限流处理方法
     * 方法签名必须与原方法一致，并增加BlockException参数
     */
    public String handleBlock(BlockException exception) {
        return "系统繁忙，请稍后再试！";
    }
}

在Sentinel Dashboard配置规则：

  1. 进入"流控规则"页面
  2. 新增规则：
    • 资源名：getOrderTest
    • 针对来源：default
    • 阈值类型：QPS
    • 单机阈值：10（每秒最多10个请求）
    • 流控模式：直接
    • 流控效果：快速失败

熔断降级 ​

Feign整合Sentinel：

application.yml配置：

yaml
spring:
  cloud:
    openfeign:
      sentinel:
        enabled: true  # 启用Sentinel对Feign的支持

降级处理类：

java
package com.example.order.feign;

import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.example.order.entity.User;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;

/**
 * Feign降级处理（Sentinel）
 */
@Component
public class UserFeignFallback implements UserFeignClient {
    
    @Override
    public User getUserById(Long id) {
        User fallbackUser = new User();
        fallbackUser.setId(id);
        fallbackUser.setUsername("降级用户");
        fallbackUser.setEmail("fallback@example.com");
        return fallbackUser;
    }
    
    @Override
    public List<User> getAllUsers() {
        return new ArrayList<>();
    }
    
    @Override
    public User createUser(User user) {
        return null;
    }
    
    @Override
    public User updateUser(Long id, User user) {
        return null;
    }
    
    @Override
    public void deleteUser(Long id) {
        // 降级处理
    }
    
    /**
     * 异常处理方法
     */
    public static User handleException(Long id, Throwable ex) {
        User fallbackUser = new User();
        fallbackUser.setId(id);
        fallbackUser.setUsername("异常降级用户");
        return fallbackUser;
    }
}

自定义降级规则：

java
package com.example.order.config;

import com.alibaba.csp.sentinel.slots.block.RuleConstant;
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRule;
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRuleManager;
import com.alibaba.csp.sentinel.slots.block.degrade.circuitbreaker.CircuitBreakerStrategy;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;

/**
 * Sentinel降级规则配置
 */
@Component
public class SentinelDegradeConfig {
    
    @PostConstruct
    public void initDegradeRules() {
        List<DegradeRule> rules = new ArrayList<>();
        
        // 慢调用比例规则
        DegradeRule slowCallRule = new DegradeRule();
        slowCallRule.setResource("getUserById");
        slowCallRule.setGrade(CircuitBreakerStrategy.SLOW_REQUEST_RATIO.getType());
        slowCallRule.setCount(0.5);  // 慢调用比例阈值（50%）
        slowCallRule.setSlowRatioThreshold(500);  // 慢调用临界值（500ms）
        slowCallRule.setTimeWindow(10);  // 熔断时长（秒）
        slowCallRule.setMinRequestAmount(5);  // 最小请求数
        slowCallRule.setStatIntervalMs(10000);  // 统计时长
        rules.add(slowCallRule);
        
        // 异常比例规则
        DegradeRule exceptionRule = new DegradeRule();
        exceptionRule.setResource("getUserById");
        exceptionRule.setGrade(CircuitBreakerStrategy.ERROR_RATIO.getType());
        exceptionRule.setCount(0.5);  // 异常比例阈值（50%）
        exceptionRule.setTimeWindow(10);
        exceptionRule.setMinRequestAmount(5);
        exceptionRule.setStatIntervalMs(10000);
        rules.add(exceptionRule);
        
        DegradeRuleManager.loadRules(rules);
    }
}

5.1.7 Spring Cloud Gateway网关 ​

Gateway简介 ​

API网关是微服务架构的统一入口，负责路由转发、鉴权、限流等功能。类似于前端的Nginx反向代理。

与前端对比：

前端Gateway
Nginx反向代理Spring Cloud Gateway
路由配置Route配置
过滤器Filter
跨域处理CORS配置

Maven依赖 ​

xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <parent>
        <groupId>com.example</groupId>
        <artifactId>microservice-parent</artifactId>
        <version>1.0.0</version>
    </parent>

    <artifactId>gateway</artifactId>
    <packaging>jar</packaging>
    <name>API网关</name>

    <dependencies>
        <!-- Spring Cloud Gateway -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-gateway</artifactId>
        </dependency>

        <!-- Nacos服务发现 -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
        </dependency>

        <!-- Spring Cloud LoadBalancer -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-loadbalancer</artifactId>
        </dependency>

        <!-- Sentinel -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>${spring-boot.version}</version>
            </plugin>
        </plugins>
    </build>
</project>

bootstrap.yml ​

yaml
spring:
  application:
    name: gateway
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848

application.yml路由配置 ​

yaml
server:
  port: 9000  # 网关端口

spring:
  cloud:
    gateway:
      # 全局跨域配置
      globalcors:
        cors-configurations:
          '[/**]':
            allowedOrigins: "*"
            allowedMethods: "*"
            allowedHeaders: "*"
            allowCredentials: false
      
      # 路由配置
      routes:
        # 用户服务路由
        - id: user-service
          uri: lb://service-user  # lb://表示负载均衡，后面是服务名
          predicates:
            - Path=/api/users/**  # 路径匹配
            - Method=GET,POST,PUT,DELETE
          filters:
            - StripPrefix=0  # 不剥离路径前缀
        
        # 订单服务路由
        - id: order-service
          uri: lb://service-order
          predicates:
            - Path=/api/orders/**
          filters:
            - StripPrefix=0
        
        # 商品服务路由（带路径剥离）
        - id: product-service
          uri: lb://service-product
          predicates:
            - Path=/product/**  # 访问/product/xxx
          filters:
            - StripPrefix=1  # 剥离第一层路径，变成/xxx
        
        # 带权重的路由
        - id: user-service-weight
          uri: lb://service-user
          predicates:
            - Weight=group1, 80  # 80%流量
          
        - id: user-service-weight-v2
          uri: lb://service-user-v2
          predicates:
            - Weight=group1, 20  # 20%流量
      
      # 默认过滤器
      default-filters:
        - AddResponseHeader=X-Gateway-Source, spring-cloud-gateway

启动类 ​

java
package com.example.gateway;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;

@SpringBootApplication
@EnableDiscoveryClient
public class GatewayApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(GatewayApplication.class, args);
        System.out.println("API网关启动成功！端口：9000");
    }
}

自定义全局过滤器 ​

鉴权过滤器：

java
package com.example.gateway.filter;

import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import java.util.Arrays;
import java.util.List;

/**
 * 全局鉴权过滤器
 */
@Component
public class AuthGlobalFilter implements GlobalFilter, Ordered {
    
    // 白名单路径
    private static final List<String> WHITE_LIST = Arrays.asList(
        "/api/users/login",
        "/api/users/register",
        "/api/products/list"
    );
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        String path = request.getURI().getPath();
        
        // 白名单直接放行
        if (WHITE_LIST.contains(path)) {
            return chain.filter(exchange);
        }
        
        // 获取Token
        String token = request.getHeaders().getFirst("Authorization");
        
        // 验证Token
        if (token == null || token.isEmpty()) {
            // 未授权，返回401
            exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
            return exchange.getResponse().setComplete();
        }
        
        // TODO: 验证Token有效性
        
        // 将用户信息传递给下游服务
        ServerHttpRequest mutatedRequest = request.mutate()
            .header("X-User-Id", "1")  // 模拟用户ID
            .header("X-User-Name", "张三")
            .build();
        
        return chain.filter(exchange.mutate().request(mutatedRequest).build());
    }
    
    /**
     * 过滤器执行顺序，数值越小优先级越高
     */
    @Override
    public int getOrder() {
        return -1;
    }
}

日志过滤器：

java
package com.example.gateway.filter;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

/**
 * 全局日志过滤器
 */
@Component
public class LoggingGlobalFilter implements GlobalFilter, Ordered {
    
    private static final Logger logger = LoggerFactory.getLogger(LoggingGlobalFilter.class);
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        
        // 请求信息
        logger.info("Gateway Request: {} {} from {}", 
            request.getMethod(),
            request.getURI().getPath(),
            request.getRemoteAddress());
        
        long startTime = System.currentTimeMillis();
        
        return chain.filter(exchange).then(Mono.fromRunnable(() -> {
            long endTime = System.currentTimeMillis();
            logger.info("Gateway Response: {} {} took {}ms",
                request.getMethod(),
                request.getURI().getPath(),
                (endTime - startTime));
        }));
    }
    
    @Override
    public int getOrder() {
        return 0;
    }
}

自定义路由工厂 ​

时间路由工厂：

java
package com.example.gateway.factory;

import org.springframework.cloud.gateway.handler.predicate.AbstractRoutePredicateFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.List;
import java.util.function.Predicate;

/**
 * 自定义路由谓词工厂：根据时间路由
 * 配置示例：
 * - TimeBetween=2024-01-01T00:00:00,2024-12-31T23:59:59
 */
@Component
public class TimeBetweenRoutePredicateFactory 
    extends AbstractRoutePredicateFactory<TimeBetweenRoutePredicateFactory.Config> {
    
    public TimeBetweenRoutePredicateFactory() {
        super(Config.class);
    }
    
    @Override
    public List<String> shortcutFieldOrder() {
        return Arrays.asList("startTime", "endTime");
    }
    
    @Override
    public Predicate<ServerWebExchange> apply(Config config) {
        return exchange -> {
            LocalDateTime now = LocalDateTime.now();
            return now.isAfter(config.getStartTime()) 
                && now.isBefore(config.getEndTime());
        };
    }
    
    public static class Config {
        private LocalDateTime startTime;
        private LocalDateTime endTime;
        
        public LocalDateTime getStartTime() {
            return startTime;
        }
        
        public void setStartTime(LocalDateTime startTime) {
            this.startTime = startTime;
        }
        
        public LocalDateTime getEndTime() {
            return endTime;
        }
        
        public void setEndTime(LocalDateTime endTime) {
            this.endTime = endTime;
        }
    }
}

5.1.8 Nacos Config配置中心 ​

配置中心简介 ​

配置中心用于集中管理微服务的配置文件，支持动态刷新。类似于前端的环境变量配置（.env文件）。

与前端对比：

前端Nacos Config
.env文件配置文件
环境变量配置项
构建时注入运行时动态获取
热更新配置动态刷新

Nacos配置管理 ​

在Nacos控制台创建配置：

  1. 进入"配置管理" -> "配置列表"
  2. 点击"+"创建配置
  3. 配置信息：
    • Data ID: service-user.yaml
    • Group: DEFAULT_GROUP
    • 配置格式: YAML
    • 配置内容:
yaml
# 数据库配置
spring:
  datasource:
    url: jdbc:mysql://localhost:3306/user_db
    username: root
    password: 123456
    driver-class-name: com.mysql.cj.jdbc.Driver

# 自定义配置
custom:
  config:
    max-retry: 3
    timeout: 5000
    feature-switch: true

动态刷新配置 ​

Controller示例：

java
package com.example.user.controller;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 配置测试控制器
 * @RefreshScope 注解使配置支持动态刷新
 */
@RestController
@RequestMapping("/api/config")
@RefreshScope
public class ConfigController {
    
    @Value("${custom.config.max-retry:3}")
    private Integer maxRetry;
    
    @Value("${custom.config.timeout:5000}")
    private Integer timeout;
    
    @Value("${custom.config.feature-switch:false}")
    private Boolean featureSwitch;
    
    @GetMapping("/test")
    public String getConfig() {
        return String.format("maxRetry: %d, timeout: %d, featureSwitch: %b",
            maxRetry, timeout, featureSwitch);
    }
}

多环境配置 ​

bootstrap.yml多环境配置：

yaml
spring:
  application:
    name: service-user
  profiles:
    active: dev  # 激活的环境
  cloud:
    nacos:
      config:
        server-addr: localhost:8848
        file-extension: yaml
        # 环境前缀
        prefix: ${spring.application.name}
        # 命名空间（用于隔离环境）
        namespace: dev-namespace-id
        # 共享配置
        shared-configs:
          - data-id: common-${spring.profiles.active}.yaml
            group: DEFAULT_GROUP
            refresh: true

Nacos配置命名规范：

${prefix}-${spring.profile.active}.${file-extension}

示例：
- service-user-dev.yaml（开发环境）
- service-user-test.yaml（测试环境）
- service-user-prod.yaml（生产环境）

5.1.9 Seata分布式事务 ​

分布式事务简介 ​

分布式事务用于保证跨多个服务的数据一致性。类似于前端的多步骤表单提交，需要保证所有步骤都成功。

与前端对比：

前端Seata
多步骤表单分布式事务
表单验证事务验证
回滚操作事务回滚
最终一致性分布式一致性

Seata简介 ​

Seata是阿里巴巴开源的分布式事务解决方案，提供AT、TCC、SAGA、XA四种事务模式。

安装Seata ​

  1. 下载Seata：Seata官网

  2. 修改配置文件：

    • conf/registry.conf：配置注册中心（Nacos）
    • conf/file.conf：配置存储模式

  3. 启动Seata Server：

bash
# Windows
bin\seata-server.bat

# Linux/Mac
bin/seata-server.sh

Maven依赖 ​

xml
<!-- Seata依赖 -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>

application.yml配置 ​

yaml
seata:
  enabled: true
  application-id: ${spring.application.name}
  tx-service-group: my_tx_group
  config:
    type: nacos
    nacos:
      server-addr: localhost:8848
      namespace: ""
      group: SEATA_GROUP
  registry:
    type: nacos
    nacos:
      server-addr: localhost:8848
      namespace: ""
      group: SEATA_GROUP
      application: default

AT模式示例 ​

订单服务创建订单：

java
package com.example.order.service;

import com.example.order.entity.Order;
import com.example.order.feign.UserFeignClient;
import io.seata.spring.annotation.GlobalTransactional;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.util.Date;

@Service
public class OrderService {
    
    @Autowired
    private UserFeignClient userFeignClient;
    
    @Autowired
    private OrderMapper orderMapper;
    
    @Autowired
    private StockFeignClient stockFeignClient;
    
    /**
     * 创建订单（分布式事务）
     * @GlobalTransactional 开启全局分布式事务
     */
    @GlobalTransactional(name = "create-order", timeoutMills = 30000)
    public Order createOrder(Order order) {
        // 1. 验证用户
        User user = userFeignClient.getUserById(order.getUserId());
        if (user == null) {
            throw new RuntimeException("用户不存在");
        }
        
        // 2. 扣减库存（调用库存服务）
        boolean stockResult = stockFeignClient.deductStock(order.getProductId(), order.getQuantity());
        if (!stockResult) {
            throw new RuntimeException("库存不足");
        }
        
        // 3. 创建订单
        order.setOrderNo("ORD" + System.currentTimeMillis());
        order.setStatus(0);
        order.setCreatedAt(new Date());
        orderMapper.insert(order);
        
        // 4. 模拟异常（测试事务回滚）
        // int i = 1 / 0;  // 抛出异常，整个事务回滚
        
        return order;
    }
}

库存服务扣减库存：

java
package com.example.stock.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class StockService {
    
    @Autowired
    private StockMapper stockMapper;
    
    /**
     * 扣减库存
     */
    public boolean deductStock(Long productId, Integer quantity) {
        Integer stock = stockMapper.getStock(productId);
        
        if (stock == null || stock < quantity) {
            return false;  // 库存不足
        }
        
        // 扣减库存
        int result = stockMapper.deductStock(productId, quantity);
        return result > 0;
    }
}

5.2 消息队列（第22-35天） ​

5.2.1 消息队列概述 ​

什么是消息队列 ​

消息队列（Message Queue，简称MQ）是一种用于异步通信的中间件，类似于前端的EventBus或发布订阅模式。

与前端对比：

前端消息队列
EventBus消息队列
发布/订阅模式消息发布/消费
异步回调异步消息处理
Promise.all消息确认
事件监听消息监听

消息队列应用场景 ​

  1. 异步处理：用户注册后发送邮件/短信
  2. 应用解耦：订单系统与库存系统解耦
  3. 流量削峰：秒杀活动中的请求排队
  4. 日志处理：集中收集和处理日志
  5. 消息通讯：聊天系统中的消息传递

主流消息队列对比 ​

特性RabbitMQKafkaRocketMQ
开发语言ErlangScala/JavaJava
协议AMQP自定义协议自定义协议
消息模型队列模型分区模型队列模型
吞吐量万级百万级十万级
延迟微秒级毫秒级毫秒级
适用场景企业应用大数据/日志金融交易

5.2.2 RabbitMQ入门 ​

RabbitMQ简介 ​

RabbitMQ是基于AMQP协议的消息队列，功能丰富，适合企业级应用。

安装RabbitMQ ​

Windows安装：

  1. 安装Erlang：Erlang官网
  2. 安装RabbitMQ：RabbitMQ官网
  3. 启用管理插件：
bash
rabbitmq-plugins enable rabbitmq_management
  1. 访问管理界面：http://localhost:15672
    • 默认账号密码：guest/guest

Maven依赖 ​

xml
<!-- RabbitMQ依赖 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml配置 ​

yaml
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /  # 虚拟主机
    # 连接配置
    connection-timeout: 10s
    # 消费者配置
    listener:
      simple:
        acknowledge-mode: manual  # 手动确认
        concurrency: 5  # 并发消费者数
        max-concurrency: 10  # 最大并发数
        prefetch: 1  # 每次只消费一条消息
      direct:
        acknowledge-mode: manual

5.2.3 RabbitMQ消息模型 ​

简单队列模型 ​

架构图：

生产者 -> 队列 -> 消费者

配置类：

java
package com.example.mq.config;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {
    
    /**
     * 创建简单队列
     */
    @Bean
    public Queue simpleQueue() {
        return new Queue("simple.queue");
    }
}

生产者：

java
package com.example.mq.producer;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class SimpleProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    /**
     * 发送消息
     */
    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend("simple.queue", message);
        System.out.println("发送消息：" + message);
    }
}

消费者：

java
package com.example.mq.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SimpleConsumer {
    
    /**
     * 监听消息
     */
    @RabbitListener(queues = "simple.queue")
    public void handleMessage(String message) {
        System.out.println("接收消息：" + message);
        // 处理业务逻辑
    }
}

工作队列模型 ​

架构图：

生产者 -> 队列 -> 消费者1
                  -> 消费者2
                  -> 消费者3

配置类：

java
@Configuration
public class RabbitMQConfig {
    
    /**
     * 工作队列（持久化）
     */
    @Bean
    public Queue workQueue() {
        return QueueBuilder.durable("work.queue").build();
    }
}

生产者：

java
@Component
public class WorkProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend("work.queue", message);
    }
    
    /**
     * 批量发送
     */
    public void sendBatchMessages(int count) {
        for (int i = 0; i < count; i++) {
            String message = "消息-" + i;
            rabbitTemplate.convertAndSend("work.queue", message);
        }
    }
}

消费者（多个实例）：

java
@Component
public class WorkConsumer {
    
    @RabbitListener(queues = "work.queue")
    public void handleMessage1(String message) {
        System.out.println("消费者1接收：" + message);
        // 模拟耗时操作
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    @RabbitListener(queues = "work.queue")
    public void handleMessage2(String message) {
        System.out.println("消费者2接收：" + message);
    }
}

发布/订阅模型 ​

架构图：

生产者 -> 交换机（Fanout） -> 队列1 -> 消费者1
                             -> 队列2 -> 消费者2
                             -> 队列3 -> 消费者3

配置类：

java
@Configuration
public class RabbitMQConfig {
    
    /**
     * 创建Fanout交换机
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanout.exchange");
    }
    
    /**
     * 创建队列
     */
    @Bean
    public Queue queue1() {
        return new Queue("fanout.queue1");
    }
    
    @Bean
    public Queue queue2() {
        return new Queue("fanout.queue2");
    }
    
    /**
     * 绑定队列到交换机
     */
    @Bean
    public Binding binding1(Queue queue1, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue1).to(fanoutExchange);
    }
    
    @Bean
    public Binding binding2(Queue queue2, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue2).to(fanoutExchange);
    }
}

生产者：

java
@Component
public class FanoutProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendMessage(String message) {
        // 发送到交换机，路由键为空（Fanout模式不需要路由键）
        rabbitTemplate.convertAndSend("fanout.exchange", "", message);
    }
}

消费者：

java
@Component
public class FanoutConsumer {
    
    @RabbitListener(queues = "fanout.queue1")
    public void handleMessage1(String message) {
        System.out.println("队列1接收：" + message);
    }
    
    @RabbitListener(queues = "fanout.queue2")
    public void handleMessage2(String message) {
        System.out.println("队列2接收：" + message);
    }
}

路由模型（Direct） ​

架构图：

生产者 -> 交换机（Direct） -> 队列1（routingKey=info） -> 消费者1
                             -> 队列2（routingKey=error） -> 消费者2

配置类：

java
@Configuration
public class RabbitMQConfig {
    
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("direct.exchange");
    }
    
    @Bean
    public Queue infoQueue() {
        return new Queue("direct.info.queue");
    }
    
    @Bean
    public Queue errorQueue() {
        return new Queue("direct.error.queue");
    }
    
    @Bean
    public Binding infoBinding(Queue infoQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(infoQueue).to(directExchange).with("info");
    }
    
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(errorQueue).to(directExchange).with("error");
    }
}

生产者：

java
@Component
public class DirectProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendInfo(String message) {
        rabbitTemplate.convertAndSend("direct.exchange", "info", message);
    }
    
    public void sendError(String message) {
        rabbitTemplate.convertAndSend("direct.exchange", "error", message);
    }
}

消费者：

java
@Component
public class DirectConsumer {
    
    @RabbitListener(queues = "direct.info.queue")
    public void handleInfo(String message) {
        System.out.println("Info消息：" + message);
    }
    
    @RabbitListener(queues = "direct.error.queue")
    public void handleError(String message) {
        System.out.println("Error消息：" + message);
    }
}

主题模型（Topic） ​

架构图：

生产者 -> 交换机（Topic） -> 队列1（routingKey=*.order.#） -> 消费者1
                             -> 队列2（routingKey=#.payment） -> 消费者2

路由键规则：

  • *：匹配一个单词
  • #：匹配零个或多个单词
  • 示例：create.order.success、order.payment

配置类：

java
@Configuration
public class RabbitMQConfig {
    
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topic.exchange");
    }
    
    @Bean
    public Queue orderQueue() {
        return new Queue("topic.order.queue");
    }
    
    @Bean
    public Queue paymentQueue() {
        return new Queue("topic.payment.queue");
    }
    
    @Bean
    public Binding orderBinding(Queue orderQueue, TopicExchange topicExchange) {
        // 匹配所有订单相关消息
        return BindingBuilder.bind(orderQueue).to(topicExchange).with("*.order.#");
    }
    
    @Bean
    public Binding paymentBinding(Queue paymentQueue, TopicExchange topicExchange) {
        // 匹配所有支付相关消息
        return BindingBuilder.bind(paymentQueue).to(topicExchange).with("#.payment");
    }
}

生产者：

java
@Component
public class TopicProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendCreateOrder(String message) {
        rabbitTemplate.convertAndSend("topic.exchange", "create.order.success", message);
    }
    
    public void sendCancelOrder(String message) {
        rabbitTemplate.convertAndSend("topic.exchange", "cancel.order", message);
    }
    
    public void sendPayment(String message) {
        rabbitTemplate.convertAndSend("topic.exchange", "order.payment.success", message);
    }
}

消费者：

java
@Component
public class TopicConsumer {
    
    @RabbitListener(queues = "topic.order.queue")
    public void handleOrder(String message, Message msg) {
        String routingKey = msg.getMessageProperties().getReceivedRoutingKey();
        System.out.println("订单消息 [" + routingKey + "]：" + message);
    }
    
    @RabbitListener(queues = "topic.payment.queue")
    public void handlePayment(String message, Message msg) {
        String routingKey = msg.getMessageProperties().getReceivedRoutingKey();
        System.out.println("支付消息 [" + routingKey + "]：" + message);
    }
}

5.2.4 消息确认机制 ​

生产者确认 ​

application.yml配置：

yaml
spring:
  rabbitmq:
    # 开启生产者确认
    publisher-confirm-type: correlated  # 异步确认
    publisher-returns: true  # 开启消息退回

生产者确认回调：

java
package com.example.mq.config;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;

@Configuration
public class RabbitMQProducerConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnsCallback(this);
    }
    
    /**
     * 确认回调（消息到达交换机）
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack) {
            System.out.println("消息成功到达交换机");
        } else {
            System.out.println("消息到达交换机失败：" + cause);
        }
    }
    
    /**
     * 退回回调（消息未路由到队列）
     */
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        System.out.println("消息被退回：" + returned.getMessage());
        System.out.println("退回原因：" + returned.getReplyText());
        System.out.println("交换机：" + returned.getExchange());
        System.out.println("路由键：" + returned.getRoutingKey());
    }
}

使用生产者确认：

java
@Component
public class ConfirmProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendMessage(String message) {
        CorrelationData correlationData = new CorrelationData();
        correlationData.getMutableContext().put("messageId", System.currentTimeMillis());
        
        rabbitTemplate.convertAndSend("direct.exchange", "info", message, correlationData);
    }
}

消费者确认 ​

手动确认消费者：

java
@Component
public class ManualAckConsumer {
    
    @RabbitListener(queues = "manual.queue")
    public void handleMessage(String message, Channel channel, 
                             @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        try {
            System.out.println("接收消息：" + message);
            
            // 处理业务逻辑
            processMessage(message);
            
            // 手动确认（成功）
            channel.basicAck(deliveryTag, false);
            
        } catch (Exception e) {
            System.out.println("处理失败：" + e.getMessage());
            
            // 手动拒绝（不重新入队）
            channel.basicNack(deliveryTag, false, false);
            
            // 或者手动拒绝（重新入队）
            // channel.basicNack(deliveryTag, false, true);
        }
    }
    
    private void processMessage(String message) throws Exception {
        // 模拟处理
        if (message.contains("error")) {
            throw new Exception("处理失败");
        }
        System.out.println("处理成功：" + message);
    }
}

5.2.5 死信队列 ​

死信队列简介 ​

死信队列（Dead Letter Queue，DLQ）用于存放无法被消费的消息。类似于前端的错误处理机制。

死信原因：

  1. 消息被拒绝（basic.reject/basic.nack）且不重新入队
  2. 消息TTL过期
  3. 队列达到最大长度

架构图：

生产者 -> 正常队列 -> 消费者
              ↓（死信）
         死信交换机 -> 死信队列 -> 死信消费者

配置类：

java
@Configuration
public class DeadLetterConfig {
    
    /**
     * 正常队列（绑定死信交换机）
     */
    @Bean
    public Queue normalQueue() {
        return QueueBuilder.durable("normal.queue")
            .deadLetterExchange("dlx.exchange")  // 死信交换机
            .deadLetterRoutingKey("dlx.routing.key")  // 死信路由键
            .ttl(10000)  // 消息TTL（10秒）
            .maxLength(100)  // 队列最大长度
            .build();
    }
    
    /**
     * 死信交换机
     */
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange("dlx.exchange");
    }
    
    /**
     * 死信队列
     */
    @Bean
    public Queue dlxQueue() {
        return QueueBuilder.durable("dlx.queue").build();
    }
    
    /**
     * 绑定死信队列到死信交换机
     */
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlx.routing.key");
    }
}

死信消费者：

java
@Component
public class DeadLetterConsumer {
    
    @RabbitListener(queues = "dlx.queue")
    public void handleDeadLetter(String message, Message msg) {
        System.out.println("===== 死信队列 =====");
        System.out.println("消息：" + message);
        System.out.println("原始交换机：" + msg.getMessageProperties().getReceivedExchange());
        System.out.println("原始路由键：" + msg.getMessageProperties().getReceivedRoutingKey());
        System.out.println("原始时间：" + msg.getMessageProperties().getReceivedTimestamp());
        System.out.println("死信原因：" + msg.getMessageProperties().getXDeathHeader());
        System.out.println("==================");
        
        // 处理死信消息（记录日志、告警等）
    }
}

5.2.6 延迟消息 ​

延迟消息简介 ​

延迟消息用于在指定时间后处理消息。类似于前端的setTimeout。

与前端对比：

前端RabbitMQ延迟消息
setTimeout延迟消息
定时器延迟队列
倒计时TTL + 死信队列

使用插件实现延迟消息 ​

安装延迟插件：

  1. 下载插件：rabbitmq_delayed_message_exchange
  2. 将插件放入RabbitMQ的plugins目录
  3. 启用插件：
bash
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

配置类：

java
@Configuration
public class DelayConfig {
    
    /**
     * 延迟交换机
     */
    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange("delay.exchange", "x-delayed-message", true, false, args);
    }
    
    /**
     * 延迟队列
     */
    @Bean
    public Queue delayQueue() {
        return QueueBuilder.durable("delay.queue").build();
    }
    
    /**
     * 绑定
     */
    @Bean
    public Binding delayBinding() {
        return BindingBuilder.bind(delayQueue()).to(delayExchange()).with("delay.routing.key").noargs();
    }
}

生产者：

java
@Component
public class DelayProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    /**
     * 发送延迟消息
     */
    public void sendDelayMessage(String message, int delayMs) {
        rabbitTemplate.convertAndSend("delay.exchange", "delay.routing.key", message, msg -> {
            msg.getMessageProperties().setDelay(delayMs);  // 设置延迟时间（毫秒）
            return msg;
        });
        System.out.println("发送延迟消息：" + message + "，延迟：" + delayMs + "ms");
    }
    
    /**
     * 订单超时取消示例
     */
    public void sendOrderTimeout(String orderNo) {
        sendDelayMessage(orderNo, 30 * 60 * 1000);  // 30分钟后取消
    }
}

消费者：

java
@Component
public class DelayConsumer {
    
    @RabbitListener(queues = "delay.queue")
    public void handleDelayMessage(String message) {
        System.out.println("处理延迟消息：" + message);
        
        // 订单超时取消
        String orderNo = message;
        cancelOrder(orderNo);
    }
    
    private void cancelOrder(String orderNo) {
        System.out.println("取消订单：" + orderNo);
        // 业务逻辑：检查订单状态，如果未支付则取消
    }
}

5.2.7 消息幂等性处理 ​

幂等性简介 ​

消息幂等性是指同一条消息被多次消费时，产生的效果与消费一次相同。类似于前端的防抖和节流。

与前端对比：

前端消息幂等性
防抖（debounce）消息去重
唯一标识消息ID
状态标记消费记录

消息幂等性实现 ​

方案1：唯一消息ID

java
@Component
public class IdempotentConsumer {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    @RabbitListener(queues = "idempotent.queue")
    public void handleMessage(String message, Message msg) throws IOException {
        String messageId = msg.getMessageProperties().getMessageId();
        System.out.println("接收消息ID：" + messageId);
        
        // 使用Redis SETNX判断是否已处理
        String key = "mq:processed:" + messageId;
        Boolean isNew = redisTemplate.opsForValue().setIfAbsent(key, "1", 24, TimeUnit.HOURS);
        
        if (Boolean.FALSE.equals(isNew)) {
            System.out.println("消息已处理，跳过：" + messageId);
            return;
        }
        
        try {
            // 处理业务逻辑
            processMessage(message);
        } catch (Exception e) {
            // 处理失败，删除标记，允许重试
            redisTemplate.delete(key);
            throw e;
        }
    }
    
    private void processMessage(String message) {
        System.out.println("处理消息：" + message);
    }
}

方案2：数据库唯一约束

java
@Component
public class DatabaseIdempotentConsumer {
    
    @Autowired
    private OrderMapper orderMapper;
    
    @RabbitListener(queues = "order.queue")
    public void handleOrderMessage(OrderMessage message) {
        try {
            // 使用订单号作为唯一标识
            Order order = new Order();
            order.setOrderNo(message.getOrderNo());
            order.setUserId(message.getUserId());
            order.setAmount(message.getAmount());
            
            // 插入数据库（订单号有唯一约束）
            orderMapper.insert(order);
            System.out.println("订单创建成功：" + message.getOrderNo());
            
        } catch (DuplicateKeyException e) {
            // 唯一约束冲突，说明消息已处理
            System.out.println("订单已存在，跳过：" + message.getOrderNo());
        }
    }
}

方案3：乐观锁（版本号）

java
@Component
public class OptimisticLockConsumer {
    
    @Autowired
    private AccountMapper accountMapper;
    
    @RabbitListener(queues = "account.queue")
    public void handleDepositMessage(DepositMessage message) {
        // 使用版本号实现乐观锁
        int result = accountMapper.depositWithVersion(
            message.getUserId(), 
            message.getAmount(),
            message.getVersion()
        );
        
        if (result > 0) {
            System.out.println("充值成功");
        } else {
            System.out.println("充值失败，版本号冲突");
        }
    }
}

Mapper XML：

xml
<update id="depositWithVersion">
    UPDATE accounts 
    SET balance = balance + #{amount},
        version = version + 1
    WHERE user_id = #{userId}
      AND version = #{version}
</update>

5.3 分布式基础（第36-42天） ​

5.3.1 CAP理论与BASE理论 ​

CAP理论 ​

CAP理论简介： CAP理论是分布式系统设计的基本理论，指出一个分布式系统最多只能同时满足以下三个特性中的两个：

  • C（Consistency，一致性）：所有节点在同一时刻看到的数据是一致的
  • A（Availability，可用性）：每个请求都能收到一个响应（不保证是最新数据）
  • P（Partition tolerance，分区容错性）：网络分区时，系统仍能继续运行

与前端对比：

前端CAP理论
数据同步（WebSocket）一致性
服务可用性可用性
网络异常处理分区容错性

CAP权衡：

  • CP（一致性 + 分区容错性）：如Zookeeper、Redis Cluster
  • AP（可用性 + 分区容错性）：如Eureka、Cassandra
  • CA（一致性 + 可用性）：单节点数据库（不常见）

BASE理论 ​

BASE理论简介： BASE理论是对CAP理论中一致性和可用性权衡的结果，核心思想是即使无法做到强一致性，但可以采用适当的方式使最终一致性得到满足。

  • BA（Basically Available，基本可用）：系统出现故障时，允许损失部分可用性
  • S（Soft state，软状态）：允许系统中的数据存在中间状态
  • E（Eventually consistent，最终一致性）：系统中的所有数据在经过一段时间后，最终能够达到一致的状态

与前端对比：

前端BASE理论
乐观更新最终一致性
加载状态软状态
降级UI基本可用

最终一致性实现方案：

  1. 读时修复
java
public User getUser(Long userId) {
    // 从缓存读取
    User user = redisTemplate.opsForValue().get("user:" + userId);
    
    if (user == null) {
        // 缓存未命中，从数据库读取
        user = userMapper.selectById(userId);
        
        if (user != null) {
            // 写入缓存
            redisTemplate.opsForValue().set("user:" + userId, user);
        }
    } else {
        // 异步检查数据一致性
        asyncCheckConsistency(userId, user);
    }
    
    return user;
}

private void asyncCheckConsistency(Long userId, User cachedUser) {
    CompletableFuture.runAsync(() -> {
        User dbUser = userMapper.selectById(userId);
        if (dbUser != null && !dbUser.equals(cachedUser)) {
            // 数据不一致，更新缓存
            redisTemplate.opsForValue().set("user:" + userId, dbUser);
        }
    });
}
  1. 写时修复
java
@Transactional
public void updateUser(User user) {
    // 更新数据库
    userMapper.updateById(user);
    
    // 删除缓存（而不是更新缓存）
    redisTemplate.delete("user:" + user.getId());
    
    // 发送消息通知其他节点更新缓存
    rabbitTemplate.convertAndSend("cache.invalidate", "user:" + user.getId());
}

5.3.2 分布式ID生成 ​

分布式ID简介 ​

分布式ID是分布式系统中用于唯一标识实体或请求的ID。类似于前端的UUID。

与前端对比：

前端分布式ID
UUID分布式ID
自增ID雪花算法
时间戳时间戳ID

分布式ID生成方案 ​

方案1：UUID

java
public class UUIDGenerator {
    
    public static String generate() {
        return UUID.randomUUID().toString();
    }
    
    public static String generateWithoutDash() {
        return UUID.randomUUID().toString().replace("-", "");
    }
    
    public static void main(String[] args) {
        System.out.println(generate());  // 550e8400-e29b-41d4-a716-446655440000
        System.out.println(generateWithoutDash());  // 550e8400e29b41d4a716446655440000
    }
}

缺点：

  • 无序，不适合作为数据库主键
  • 长度较长，占用空间大

方案2：数据库自增ID

java
public class DatabaseIdGenerator {
    
    @Autowired
    private IdMapper idMapper;
    
    public Long generate() {
        // 使用数据库自增ID
        IdEntity entity = new IdEntity();
        idMapper.insert(entity);
        return entity.getId();
    }
}

缺点：

  • 依赖数据库，性能瓶颈
  • 单点故障

方案3：雪花算法（Snowflake）

雪花算法简介： 雪花算法是Twitter开源的分布式ID生成算法，生成的ID是一个64位的long型数字。

ID结构：

0 - 41位时间戳 - 10位工作机器ID - 12位序列号

- 1位符号位：始终为0
- 41位时间戳：毫秒级，可使用69年
- 10位工作机器ID：支持1024个节点
- 12位序列号：每毫秒可生成4096个ID

雪花算法实现：

java
package com.example.common.id;

public class SnowflakeIdGenerator {
    
    // 起始时间戳（2020-01-01）
    private final long epoch = 1577836800000L;
    
    // 机器ID位数
    private final long workerIdBits = 5L;
    
    // 数据中心ID位数
    private final long datacenterIdBits = 5L;
    
    // 序列号位数
    private final long sequenceBits = 12L;
    
    // 机器ID最大值
    private final long maxWorkerId = ~(-1L << workerIdBits);  // 31
    
    // 数据中心ID最大值
    private final long maxDatacenterId = ~(-1L << datacenterIdBits);  // 31
    
    // 机器ID左移位数
    private final long workerIdShift = sequenceBits;  // 12
    
    // 数据中心ID左移位数
    private final long datacenterIdShift = sequenceBits + workerIdBits;  // 17
    
    // 时间戳左移位数
    private final long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;  // 22
    
    // 序列号掩码
    private final long sequenceMask = ~(-1L << sequenceBits);  // 4095
    
    // 工作机器ID（0-31）
    private long workerId;
    
    // 数据中心ID（0-31）
    private long datacenterId;
    
    // 序列号（0-4095）
    private long sequence = 0L;
    
    // 上次生成ID的时间戳
    private long lastTimestamp = -1L;
    
    public SnowflakeIdGenerator(long workerId, long datacenterId) {
        if (workerId > maxWorkerId || workerId < 0) {
            throw new IllegalArgumentException("Worker ID can't be greater than " + maxWorkerId + " or less than 0");
        }
        if (datacenterId > maxDatacenterId || datacenterId < 0) {
            throw new IllegalArgumentException("Datacenter ID can't be greater than " + maxDatacenterId + " or less than 0");
        }
        this.workerId = workerId;
        this.datacenterId = datacenterId;
    }
    
    /**
     * 生成下一个ID
     */
    public synchronized long nextId() {
        long timestamp = System.currentTimeMillis();
        
        if (timestamp < lastTimestamp) {
            throw new RuntimeException("Clock moved backwards. Refusing to generate id for " + 
                (lastTimestamp - timestamp) + " milliseconds");
        }
        
        if (lastTimestamp == timestamp) {
            // 同一毫秒，递增序列号
            sequence = (sequence + 1) & sequenceMask;
            if (sequence == 0) {
                // 序列号溢出，等待下一毫秒
                timestamp = tilNextMillis(lastTimestamp);
            }
        } else {
            // 新的毫秒，序列号重置为0
            sequence = 0L;
        }
        
        lastTimestamp = timestamp;
        
        // 组合ID
        return ((timestamp - epoch) << timestampLeftShift) |
               (datacenterId << datacenterIdShift) |
               (workerId << workerIdShift) |
               sequence;
    }
    
    /**
     * 等待下一毫秒
     */
    private long tilNextMillis(long lastTimestamp) {
        long timestamp = System.currentTimeMillis();
        while (timestamp <= lastTimestamp) {
            timestamp = System.currentTimeMillis();
        }
        return timestamp;
    }
    
    /**
     * 解析ID
     */
    public void parseId(long id) {
        long timestamp = (id >> timestampLeftShift) + epoch;
        long datacenterId = (id >> datacenterIdShift) & maxDatacenterId;
        long workerId = (id >> workerIdShift) & maxWorkerId;
        long sequence = id & sequenceMask;
        
        System.out.println("时间戳：" + timestamp);
        System.out.println("数据中心ID：" + datacenterId);
        System.out.println("机器ID：" + workerId);
        System.out.println("序列号：" + sequence);
    }
    
    public static void main(String[] args) {
        SnowflakeIdGenerator generator = new SnowflakeIdGenerator(1, 1);
        
        for (int i = 0; i < 10; i++) {
            long id = generator.nextId();
            System.out.println(id);
        }
    }
}

Spring Boot集成雪花算法：

java
package com.example.common.config;

import com.example.common.id.SnowflakeIdGenerator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class IdGeneratorConfig {
    
    @Bean
    public SnowflakeIdGenerator snowflakeIdGenerator() {
        // 从配置文件读取workerId和datacenterId
        return new SnowflakeIdGenerator(1, 1);
    }
}

使用示例：

java
@Service
public class OrderService {
    
    @Autowired
    private SnowflakeIdGenerator idGenerator;
    
    public Order createOrder(Order order) {
        // 生成分布式ID
        order.setId(idGenerator.nextId());
        order.setOrderNo("ORD" + idGenerator.nextId());
        
        // 保存订单
        orderMapper.insert(order);
        
        return order;
    }
}

5.3.3 分布式锁 ​

分布式锁简介 ​

分布式锁是用于在分布式系统中控制对共享资源的访问。类似于前端的互斥锁（Mutex）。

与前端对比：

前端分布式锁
Mutex分布式锁
乐观锁（版本号）乐观锁
悲观锁（Lock）悲观锁

Redis实现分布式锁 ​

方案1：基础实现

java
@Component
public class RedisLock {
    
    @Autowired
    private StringRedisTemplate redisTemplate;
    
    /**
     * 获取锁
     */
    public boolean lock(String key, String value, long expireSeconds) {
        // SETNX：如果key不存在，则设置成功返回true
        Boolean result = redisTemplate.opsForValue()
            .setIfAbsent(key, value, expireSeconds, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(result);
    }
    
    /**
     * 释放锁
     */
    public void unlock(String key, String value) {
        // 只释放自己持有的锁
        String currentValue = redisTemplate.opsForValue().get(key);
        if (value.equals(currentValue)) {
            redisTemplate.delete(key);
        }
    }
}

使用示例：

java
@Service
public class StockService {
    
    @Autowired
    private RedisLock redisLock;
    
    @Autowired
    private StockMapper stockMapper;
    
    /**
     * 扣减库存（加锁）
     */
    public boolean deductStock(Long productId, Integer quantity) {
        String lockKey = "lock:stock:" + productId;
        String lockValue = UUID.randomUUID().toString();
        
        try {
            // 获取锁
            boolean locked = redisLock.lock(lockKey, lockValue, 10);
            if (!locked) {
                System.out.println("获取锁失败，请稍后重试");
                return false;
            }
            
            // 执行业务逻辑
            Integer stock = stockMapper.getStock(productId);
            if (stock == null || stock < quantity) {
                return false;
            }
            
            // 扣减库存
            stockMapper.deductStock(productId, quantity);
            return true;
            
        } finally {
            // 释放锁
            redisLock.unlock(lockKey, lockValue);
        }
    }
}

方案2：Redisson分布式锁（推荐）

Maven依赖：

xml
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.24.3</version>
</dependency>

application.yml配置：

yaml
spring:
  redis:
    host: localhost
    port: 6379
    password:
    database: 0

使用Redisson：

java
@Service
public class StockService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    @Autowired
    private StockMapper stockMapper;
    
    /**
     * 扣减库存（Redisson分布式锁）
     */
    public boolean deductStock(Long productId, Integer quantity) {
        String lockKey = "lock:stock:" + productId;
        
        // 获取锁对象
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 尝试获取锁（等待3秒，锁自动释放时间10秒）
            boolean locked = lock.tryLock(3, 10, TimeUnit.SECONDS);
            if (!locked) {
                System.out.println("获取锁失败");
                return false;
            }
            
            // 执行业务逻辑
            Integer stock = stockMapper.getStock(productId);
            if (stock == null || stock < quantity) {
                return false;
            }
            
            stockMapper.deductStock(productId, quantity);
            return true;
            
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        } finally {
            // 释放锁（只释放自己持有的锁）
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
    
    /**
     * 使用注解方式（更简洁）
     */
    @RedissonLock(key = "'stock:' + #productId", expireSeconds = 10)
    public boolean deductStockWithAnnotation(Long productId, Integer quantity) {
        Integer stock = stockMapper.getStock(productId);
        if (stock == null || stock < quantity) {
            return false;
        }
        return stockMapper.deductStock(productId, quantity) > 0;
    }
}

自定义分布式锁注解：

java
package com.example.common.annotation;

import java.lang.annotation.*;

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RedissonLock {
    
    /**
     * 锁的key（支持SpEL表达式）
     */
    String key();
    
    /**
     * 锁过期时间（秒）
     */
    long expireSeconds() default 30;
    
    /**
     * 获取锁超时时间（秒）
     */
    long timeoutSeconds() default 3;
    
    /**
     * 错误提示信息
     */
    String errorMessage() default "系统繁忙，请稍后重试";
}

AOP切面实现：

java
package com.example.common.aspect;

import com.example.common.annotation.RedissonLock;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.DefaultParameterNameDiscoverer;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;

@Aspect
@Component
public class RedissonLockAspect {
    
    @Autowired
    private RedissonClient redissonClient;
    
    private SpelExpressionParser parser = new SpelExpressionParser();
    private DefaultParameterNameDiscoverer nameDiscoverer = new DefaultParameterNameDiscoverer();
    
    @Around("@annotation(redissonLock)")
    public Object around(ProceedingJoinPoint joinPoint, RedissonLock redissonLock) throws Throwable {
        // 解析SpEL表达式
        String key = parseKey(redissonLock.key(), joinPoint);
        String lockKey = "lock:" + key;
        
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 获取锁
            boolean locked = lock.tryLock(
                redissonLock.timeoutSeconds(),
                redissonLock.expireSeconds(),
                TimeUnit.SECONDS
            );
            
            if (!locked) {
                throw new RuntimeException(redissonLock.errorMessage());
            }
            
            // 执行方法
            return joinPoint.proceed();
            
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
    
    /**
     * 解析SpEL表达式
     */
    private String parseKey(String keyExpression, ProceedingJoinPoint joinPoint) {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        
        EvaluationContext context = new StandardEvaluationContext();
        String[] paramNames = nameDiscoverer.getParameterNames(method);
        Object[] args = joinPoint.getArgs();
        
        if (paramNames != null) {
            for (int i = 0; i < paramNames.length; i++) {
                context.setVariable(paramNames[i], args[i]);
            }
        }
        
        Expression expression = parser.parseExpression(keyExpression);
        return expression.getValue(context, String.class);
    }
}

Zookeeper实现分布式锁 ​

Maven依赖：

xml
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.5.0</version>
</dependency>

使用Curator：

java
@Service
public class ZookeeperLockService {
    
    @Autowired
    private CuratorFramework curatorFramework;
    
    /**
     * 使用Zookeeper实现分布式锁
     */
    public void executeWithLock(String lockPath, Runnable task) throws Exception {
        // 创建锁
        InterProcessMutex lock = new InterProcessMutex(curatorFramework, lockPath);
        
        try {
            // 获取锁（等待5秒）
            boolean acquired = lock.acquire(5, TimeUnit.SECONDS);
            if (!acquired) {
                throw new RuntimeException("获取锁失败");
            }
            
            // 执行业务逻辑
            task.run();
            
        } finally {
            // 释放锁
            lock.release();
        }
    }
}

5.3.4 分布式会话管理 ​

分布式会话简介 ​

分布式会话用于在多个服务实例之间共享用户会话信息。类似于前端的全局状态管理（Redux/Vuex）。

与前端对比：

前端分布式会话
LocalStorage分布式会话
SessionStorage会话存储
Redux/Vuex共享状态

Session共享方案 ​

方案1：Spring Session + Redis

Maven依赖：

xml
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.session</groupId>
    <artifactId>spring-session-data-redis</artifactId>
</dependency>

application.yml配置：

yaml
spring:
  session:
    store-type: redis  # 使用Redis存储Session
    timeout: 1800s  # Session超时时间（30分钟）
    redis:
      namespace: "session:"  # Redis命名空间

配置类：

java
@Configuration
@EnableRedisHttpSession
public class SessionConfig {
    
    /**
     * 自定义Redis序列化器
     */
    @Bean
    public RedisSerializer<Object> springSessionDefaultRedisSerializer() {
        return new GenericJackson2JsonRedisSerializer();
    }
}

使用示例：

java
@RestController
@RequestMapping("/api/auth")
public class AuthController {
    
    @Autowired
    private HttpServletRequest request;
    
    /**
     * 登录
     */
    @PostMapping("/login")
    public String login(@RequestBody LoginDTO loginDTO) {
        // 验证用户名密码
        User user = userService.authenticate(loginDTO.getUsername(), loginDTO.getPassword());
        
        if (user != null) {
            // 将用户信息存入Session
            request.getSession().setAttribute("currentUser", user);
            return "登录成功";
        }
        
        return "登录失败";
    }
    
    /**
     * 获取当前用户
     */
    @GetMapping("/current")
    public User getCurrentUser() {
        return (User) request.getSession().getAttribute("currentUser");
    }
    
    /**
     * 登出
     */
    @PostMapping("/logout")
    public String logout() {
        request.getSession().invalidate();
        return "登出成功";
    }
}

方案2：JWT Token

Maven依赖：

xml
<dependency>
    <groupId>io.jsonwebtoken</groupId>
    <artifactId>jjwt-api</artifactId>
    <version>0.11.5</version>
</dependency>
<dependency>
    <groupId>io.jsonwebtoken</groupId>
    <artifactId>jjwt-impl</artifactId>
    <version>0.11.5</version>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>io.jsonwebtoken</groupId>
    <artifactId>jjwt-jackson</artifactId>
    <version>0.11.5</version>
    <scope>runtime</scope>
</dependency>

JWT工具类：

java
package com.example.common.util;

import io.jsonwebtoken.*;
import io.jsonwebtoken.security.Keys;
import java.security.Key;
import java.util.Date;
import java.util.Map;

public class JwtUtil {
    
    // 密钥（生产环境应使用更安全的密钥）
    private static final Key KEY = Keys.secretKeyFor(SignatureAlgorithm.HS256);
    
    // 过期时间（24小时）
    private static final long EXPIRATION = 24 * 60 * 60 * 1000;
    
    /**
     * 生成Token
     */
    public static String generateToken(String userId, Map<String, Object> claims) {
        return Jwts.builder()
            .setClaims(claims)
            .setSubject(userId)
            .setIssuedAt(new Date())
            .setExpiration(new Date(System.currentTimeMillis() + EXPIRATION))
            .signWith(KEY)
            .compact();
    }
    
    /**
     * 解析Token
     */
    public static Claims parseToken(String token) {
        return Jwts.parserBuilder()
            .setSigningKey(KEY)
            .build()
            .parseClaimsJws(token)
            .getBody();
    }
    
    /**
     * 验证Token
     */
    public static boolean validateToken(String token) {
        try {
            parseToken(token);
            return true;
        } catch (JwtException | IllegalArgumentException e) {
            return false;
        }
    }
    
    /**
     * 获取用户ID
     */
    public static String getUserId(String token) {
        return parseToken(token).getSubject();
    }
    
    /**
     * 刷新Token
     */
    public static String refreshToken(String token) {
        Claims claims = parseToken(token);
        claims.setIssuedAt(new Date());
        claims.setExpiration(new Date(System.currentTimeMillis() + EXPIRATION));
        
        return Jwts.builder()
            .setClaims(claims)
            .signWith(KEY)
            .compact();
    }
}

JWT拦截器：

java
package com.example.gateway.filter;

import com.example.common.util.JwtUtil;
import io.jsonwebtoken.Claims;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import java.util.Arrays;
import java.util.List;

@Component
public class JwtAuthFilter implements GlobalFilter, Ordered {
    
    private static final List<String> WHITE_LIST = Arrays.asList(
        "/api/auth/login",
        "/api/auth/register"
    );
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        String path = exchange.getRequest().getURI().getPath();
        
        // 白名单放行
        if (WHITE_LIST.contains(path)) {
            return chain.filter(exchange);
        }
        
        // 获取Token
        String token = exchange.getRequest().getHeaders().getFirst("Authorization");
        
        if (token == null || token.isEmpty()) {
            exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
            return exchange.getResponse().setComplete();
        }
        
        // 验证Token
        if (!JwtUtil.validateToken(token)) {
            exchange.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED);
            return exchange.getResponse().setComplete();
        }
        
        // 解析用户信息
        Claims claims = JwtUtil.parseToken(token);
        String userId = claims.getSubject();
        
        // 将用户信息传递给下游服务
        ServerHttpRequest mutatedRequest = exchange.getRequest().mutate()
            .header("X-User-Id", userId)
            .build();
        
        return chain.filter(exchange.mutate().request(mutatedRequest).build());
    }
    
    @Override
    public int getOrder() {
        return -1;
    }
}

5.3.5 分库分表（ShardingSphere） ​

分库分表简介 ​

分库分表是用于解决单库单表数据量过大导致的性能问题。类似于前端的数据分页和虚拟滚动。

与前端对比：

前端分库分表
数据分页分表
数据分片分库
虚拟滚动数据路由

为什么要分库分表 ​

  • 单表数据量过大（超过1000万行）
  • 数据库连接数瓶颈
  • 读写性能瓶颈
  • 单库容量限制

分库分表策略 ​

垂直拆分：

  • 按业务拆分：用户库、订单库、商品库
  • 按字段拆分：大字段单独存储

水平拆分：

  • 按ID取模：user_id % 4
  • 按时间范围：按月/年拆分
  • 按哈希范围：一致性哈希

ShardingSphere简介 ​

ShardingSphere是Apache开源的分布式数据库中间件，提供分库分表、读写分离等功能。

Maven依赖 ​

xml
<dependency>
    <groupId>org.apache.shardingsphere</groupId>
    <artifactId>shardingsphere-jdbc-core-spring-boot-starter</artifactId>
    <version>5.3.2</version>
</dependency>

application.yml配置 ​

分表配置：

yaml
spring:
  shardingsphere:
    # 数据源配置
    datasource:
      names: ds0,ds1
      ds0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/db_0
        username: root
        password: 123456
      ds1:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3307/db_1
        username: root
        password: 123456
    
    # 分片规则配置
    rules:
      sharding:
        # 分片表配置
        tables:
          # 订单表
          t_order:
            # 实际数据节点
            actual-data-nodes: ds$->{0..1}.t_order_$->{0..1}
            # 分库策略
            database-strategy:
              standard:
                sharding-column: user_id
                sharding-algorithm-name: db-hash-mod
            # 分表策略
            table-strategy:
              standard:
                sharding-column: order_id
                sharding-algorithm-name: table-hash-mod
        
        # 分片算法
        sharding-algorithms:
          # 分库算法（按user_id取模）
          db-hash-mod:
            type: HASH_MOD
            props:
              sharding-count: 2
          # 分表算法（按order_id取模）
          table-hash-mod:
            type: HASH_MOD
            props:
              sharding-count: 2
    
    # 分布式ID生成器
    props:
      sql-show: true  # 打印SQL

分表配置（单库多表）：

yaml
spring:
  shardingsphere:
    datasource:
      names: ds0
      ds0:
        type: com.zaxxer.hikari.HikariDataSource
        driver-class-name: com.mysql.cj.jdbc.Driver
        jdbc-url: jdbc:mysql://localhost:3306/sharding_db
        username: root
        password: 123456
    
    rules:
      sharding:
        tables:
          # 用户表（分4张表）
          t_user:
            actual-data-nodes: ds0.t_user_$->{0..3}
            table-strategy:
              standard:
                sharding-column: id
                sharding-algorithm-name: user-mod
        
        sharding-algorithms:
          user-mod:
            type: MOD
            props:
              sharding-count: 4

分表实体类 ​

java
package com.example.order.entity;

import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

@Data
@TableName("t_order")
public class Order {
    private Long orderId;
    private Long userId;
    private String orderNo;
    private BigDecimal amount;
    private Integer status;
    private Date createdAt;
}

使用示例 ​

java
@Service
public class OrderService {
    
    @Autowired
    private OrderMapper orderMapper;
    
    /**
     * 创建订单（自动路由到对应的库表）
     */
    public Order createOrder(Order order) {
        order.setOrderId(snowflakeIdGenerator.nextId());
        order.setOrderNo("ORD" + System.currentTimeMillis());
        order.setCreatedAt(new Date());
        
        // ShardingSphere会自动根据user_id和order_id路由到对应的库表
        orderMapper.insert(order);
        
        return order;
    }
    
    /**
     * 查询订单（需要包含分片键）
     */
    public Order getOrderById(Long userId, Long orderId) {
        // 查询时必须包含分片键（user_id），否则会全路由
        return orderMapper.selectById(orderId);
    }
}

分库分表注意事项 ​

  1. 分片键选择

    • 选择查询频率高的字段
    • 选择数据分布均匀的字段
    • 避免使用会频繁更新的字段

  2. 跨分片查询

    • 尽量避免跨分片查询
    • 使用冗余字段或宽表
    • 使用ES等搜索引擎

  3. 分布式事务

    • 使用Seata保证跨库事务
    • 使用消息队列实现最终一致性

  4. 数据迁移

    • 使用双写方案
    • 使用ShardingSphere的数据迁移工具

5.4 练习任务 ​

练习1：微服务电商系统 ​

  • 搭建完整的微服务架构（用户服务、商品服务、订单服务）
  • 使用Nacos作为服务注册与发现
  • 使用OpenFeign实现服务间调用
  • 使用Sentinel实现熔断降级
  • 使用Gateway实现统一入口
  • 使用RabbitMQ实现订单异步处理
  • 使用Seata实现分布式事务

练习2：秒杀系统 ​

  • 实现秒杀接口（限流、防刷）
  • 使用Redis预减库存
  • 使用RabbitMQ异步下单
  • 使用分布式锁防止超卖
  • 使用Sentinel实现服务降级

练习3：分布式日志系统 ​

  • 使用RabbitMQ收集日志
  • 实现日志分级处理（INFO/WARN/ERROR）
  • 使用死信队列处理异常日志
  • 实现日志持久化到数据库
  • 实现日志查询接口

练习4：分布式ID生成服务 ​

  • 实现雪花算法生成ID
  • 提供REST API生成ID
  • 支持批量生成ID
  • 实现ID解析功能
  • 集成到订单系统中

5.5 学习检查清单 ​

  • [ ] 理解微服务架构与单体架构的区别
  • [ ] 掌握Spring Cloud Alibaba核心组件
  • [ ] 能够搭建完整的微服务架构
  • [ ] 熟练使用Nacos进行服务注册与发现
  • [ ] 熟练使用OpenFeign进行服务调用
  • [ ] 理解负载均衡的原理和配置
  • [ ] 掌握Sentinel的流量控制和熔断降级
  • [ ] 熟练使用Spring Cloud Gateway
  • [ ] 理解配置中心的作用和使用
  • [ ] 掌握分布式事务的概念和实现
  • [ ] 理解消息队列的应用场景
  • [ ] 熟练使用RabbitMQ的各种消息模型
  • [ ] 掌握消息确认和死信队列
  • [ ] 理解延迟消息的实现
  • [ ] 掌握消息幂等性的处理方法
  • [ ] 理解CAP理论和BASE理论
  • [ ] 掌握分布式ID的生成方案
  • [ ] 熟练使用Redis实现分布式锁
  • [ ] 理解分布式会话管理的方案
  • [ ] 掌握分库分表的基本概念和实现

完成本阶段后，你应该能够搭建完整的微服务架构，理解分布式系统的基本概念和实现方案，为实际项目开发打下基础。

评论

正在加载评论…