【阶段 6】消息与异步：RabbitMQ、Kafka 与异步任务

# 1. RabbitMQ 3.12（带管理界面，管理端口 15672）
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 
  rabbitmq:3.12-management

# 2. Kafka 3.6（KRaft 单节点模式，无需 ZooKeeper）
docker run -d --name kafka -p 9092:9092 
  -e KAFKA_CFG_NODE_ID=0 
  -e KAFKA_CFG_PROCESS_ROLES=controller,broker 
  -e KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093 
  -e KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 
  -e KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 
  -e KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER 
  -e KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT 
  bitnami/kafka:3.6

# 3. Redis 7
docker run -d --name redis -p 6379:6379 redis:7

# 4. MySQL 8.0（密码 root / root，库名 demo）
docker run -d --name mysql -p 3306:3306 
  -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=demo mysql:8.0

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.2.5</version>
</parent>
<properties>
    <java.version>17</java.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-spring-boot3-starter</artifactId>
        <version>3.5.7</version>
    </dependency>
    <dependency>
        <groupId>com.mysql</groupId>
        <artifactId>mysql-connector-j</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <!-- fastjson2：消息体 JSON 序列化（本地消息表 payload、事件对象） -->
    <dependency>
        <groupId>com.alibaba.fastjson2</groupId>
        <artifactId>fastjson2</artifactId>
        <version>2.0.51</version>
    </dependency>
</dependencies>

package com.example.demo.config;

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

/**
 * 三种交换机的最小演示配置。
 * 目的：让读者直观理解 direct / fanout / topic 的路由差异。
 */
@Configuration
public class RabbitExchangeDemoConfig {

    public static final String DIRECT_EXCHANGE = "demo.direct";
    public static final String FANOUT_EXCHANGE = "demo.fanout";
    public static final String TOPIC_EXCHANGE = "demo.topic";

    // ---------- direct：精确匹配 ----------
    @Bean
    public DirectExchange directExchange() {
        // durable=true 表示交换机持久化，Broker 重启后仍存在
        return ExchangeBuilder.directExchange(DIRECT_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue directQueueA() {
        // 队列也持久化，避免重启丢队列
        return QueueBuilder.durable("demo.direct.queue.a").build();
    }

    @Bean
    public Binding directBindingA() {
        // 只有 routingKey 精确等于 "order.created" 的消息才会进入该队列
        return BindingBuilder.bind(directQueueA()).to(directExchange()).with("order.created");
    }

    // ---------- fanout：广播 ----------
    @Bean
    public FanoutExchange fanoutExchange() {
        return ExchangeBuilder.fanoutExchange(FANOUT_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue fanoutQueueA() {
        return QueueBuilder.durable("demo.fanout.queue.a").build();
    }

    @Bean
    public Queue fanoutQueueB() {
        return QueueBuilder.durable("demo.fanout.queue.b").build();
    }

    @Bean
    public Binding fanoutBindingA() {
        // fanout 忽略 routingKey，所有绑定队列都会收到消息
        return BindingBuilder.bind(fanoutQueueA()).to(fanoutExchange());
    }

    @Bean
    public Binding fanoutBindingB() {
        return BindingBuilder.bind(fanoutQueueB()).to(fanoutExchange());
    }

    // ---------- topic：通配符匹配 ----------
    @Bean
    public TopicExchange topicExchange() {
        return ExchangeBuilder.topicExchange(TOPIC_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue topicQueueOrder() {
        return QueueBuilder.durable("demo.topic.queue.order").build();
    }

    @Bean
    public Queue topicQueueAll() {
        return QueueBuilder.durable("demo.topic.queue.all").build();
    }

    @Bean
    public Binding topicBindingOrder() {
        // "order.*" 匹配 order.created、order.paid 等两段式 routingKey
        return BindingBuilder.bind(topicQueueOrder()).to(topicExchange()).with("order.*");
    }

    @Bean
    public Binding topicBindingAll() {
        // "order.#" 匹配以 order. 开头的任意多段 routingKey（含 order.created.sub）
        return BindingBuilder.bind(topicQueueAll()).to(topicExchange()).with("order.#");
    }
}

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    # 生产者确认：correlated 表示每条消息都有独立的 confirm 回调
    publisher-confirm-type: correlated
    # 生产者退回：消息路由不到任何队列时，回调 ReturnCallback
    publisher-returns: true
    template:
      # mandatory=true 时，路由失败的消息才会触发 ReturnCallback
      mandatory: true
    listener:
      simple:
        # 手动 ACK：消费成功后由代码显式 ack，失败 nack，避免自动 ack 造成「处理失败也删消息」
        acknowledge-mode: manual
        # 预取数量：每个消费者最多同时取 10 条未 ack 消息，防止消费端被冲垮
        prefetch: 10
        # 消费失败时的重试（配合手动 ACK 使用，见 2.6 关键点）
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000
          multiplier: 2.0

package com.example.demo.config;

import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitTemplate 统一配置：使用 JSON 序列化。
 * 原因：默认的 Java 序列化字节流，跨语言无法消费，且可读性差、体积大。
 */
@Configuration
public class RabbitTemplateConfig {

    @Bean
    public MessageConverter jacksonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        // 发送端使用 JSON 转换器，消息体以 JSON 形式存储
        template.setMessageConverter(jacksonMessageConverter());
        return template;
    }
}

package com.example.demo.mq;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

/**
 * 订单消息生产者（生产级）。
 * 通过 confirm + return 双回调，把「消息到底发没发出去」这件事显式化。
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderMessageProducer {

    private final RabbitTemplate rabbitTemplate;

    /**
     * 发送「订单已创建」事件。
     *
     * @param event 订单事件体
     * @param bizId 业务唯一 id，用于 confirm 回调定位是哪条消息出了问题
     */
    public void sendOrderCreatedEvent(OrderCreatedEvent event, String bizId) {
        // CorrelationData 是本次发送的「凭据」，confirm 回调里通过它关联具体消息
        CorrelationData correlationData = new CorrelationData(bizId);

        // 注意：RabbitTemplate 只发送，不感知结果；结果在下面两个回调里异步返回
        rabbitTemplate.convertAndSend(
                RabbitOrderConfig.ORDER_EXCHANGE,
                RabbitOrderConfig.ORDER_CREATED_ROUTING_KEY,
                event,
                correlationData
        );
        log.info("已发送订单创建消息，bizId={}, orderId={}", bizId, event.getOrderId());
    }
}

package com.example.demo.config;

import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;

/**
 * 注册生产者的 confirm 与 return 回调。
 * 这是「消息不丢」的第一道防线：发出去了必须知道结果。
 */
@Slf4j
@Configuration
@RequiredArgsConstructor
public class RabbitReliabilityConfig {

    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void registerCallbacks() {
        // 1. 确认回调：Broker 是否成功接收（到达交换机即回调，与路由无关）
        rabbitTemplate.setConfirmCallback((CorrelationData data, boolean ack, String cause) -> {
            if (ack) {
                log.info("消息发送确认成功，bizId={}", data == null ? null : data.getId());
            } else {
                // 生产上这里应：告警 + 记录待补偿 + 定时重发或人工介入
                log.error("消息发送确认失败！bizId={}, cause={}",
                        data == null ? null : data.getId(), cause);
            }
        });

        // 2. 退回回调：交换机收到了，但路由不到任何队列（routingKey/绑定配错）
        rabbitTemplate.setReturnsCallback((ReturnedMessage returned) -> {
            Message msg = returned.getMessage();
            log.error("消息路由失败：exchange={}, routingKey={}, replyText={}, body={}",
                    returned.getExchange(), returned.getRoutingKey(),
                    returned.getReplyText(), new String(msg.getBody()));
        });
    }
}

package com.example.demo.mq;

import com.example.demo.service.StockService;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 订单消息消费者（生产级）。
 * 手动 ACK + 幂等，是「消息不丢、不重」的关键。
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderMessageConsumer {

    private final StockService stockService;

    @RabbitListener(queues = RabbitOrderConfig.ORDER_CREATED_QUEUE)
    public void onOrderCreated(OrderCreatedEvent event, Message message, Channel channel,
                               @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        log.info("收到订单创建消息，orderId={}, deliveryTag={}", event.getOrderId(), deliveryTag);
        try {
            // 扣库存内部必须做幂等（唯一键 / Redis setNX），保证重复投递不会重复扣
            stockService.deductStock(event);
            // 处理成功：手动 ack，multiple=false 表示只确认当前这一条
            channel.basicAck(deliveryTag, false);
            log.info("订单创建消息处理成功，orderId={}", event.getOrderId());
        } catch (Exception e) {
            log.error("订单创建消息处理失败，orderId={}", event.getOrderId(), e);
            // 失败：不重回队列（requeue=false），消息进入死信队列等待人工/定时补偿
            // 若希望立即重试，可改为 requeue=true，但需配合消费幂等 + 重试上限，否则会死循环
            channel.basicNack(deliveryTag, false, false);
        }
    }
}

package com.example.demo.config;

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

/**
 * 订单业务队列 + 延迟关单队列（死信队列模式）。
 * 主线：三大可靠性问题之「消息丢失」——通过持久化 + 手动 ACK + 死信兜底解决。
 */
@Configuration
public class RabbitOrderConfig {

    // ---------- 正常业务：订单创建 ----------
    public static final String ORDER_EXCHANGE = "order.exchange";
    public static final String ORDER_CREATED_QUEUE = "order.created.queue";
    public static final String ORDER_CREATED_ROUTING_KEY = "order.created";

    // ---------- 延迟关单：30 分钟未支付自动关闭 ----------
    public static final String ORDER_DELAY_QUEUE = "order.delay.queue";      // 延迟队列（无消费者）
    public static final String ORDER_DLX_EXCHANGE = "order.dlx";             // 死信交换机
    public static final String ORDER_DLX_QUEUE = "order.dlx.queue";          // 死信队列（有消费者）
    public static final String ORDER_DLX_ROUTING_KEY = "order.close";

    /** 订单创建交换机 */
    @Bean
    public DirectExchange orderExchange() {
        return ExchangeBuilder.directExchange(ORDER_EXCHANGE).durable(true).build();
    }

    /** 订单创建队列：供库存/通知服务消费 */
    @Bean
    public Queue orderCreatedQueue() {
        return QueueBuilder.durable(ORDER_CREATED_QUEUE).build();
    }

    @Bean
    public Binding orderCreatedBinding() {
        return BindingBuilder.bind(orderCreatedQueue())
                .to(orderExchange()).with(ORDER_CREATED_ROUTING_KEY);
    }

    /** 延迟队列：TTL 30 分钟，且不绑定任何消费者，超时后消息自动变死信 */
    @Bean
    public Queue orderDelayQueue() {
        return QueueBuilder.durable(ORDER_DELAY_QUEUE)
                // 消息在队列里的最长存活时间：30 分钟，超时即成为死信
                .ttl(30 * 60 * 1000)
                // 指定死信交换机与死信路由键，死信会按此路由
                .deadLetterExchange(ORDER_DLX_EXCHANGE)
                .deadLetterRoutingKey(ORDER_DLX_ROUTING_KEY)
                .build();
    }

    /** 死信交换机 */
    @Bean
    public DirectExchange orderDlxExchange() {
        return ExchangeBuilder.directExchange(ORDER_DLX_EXCHANGE).durable(true).build();
    }

    /** 死信队列：真正执行「关单」的消费者监听这里 */
    @Bean
    public Queue orderDlxQueue() {
        return QueueBuilder.durable(ORDER_DLX_QUEUE).build();
    }

    @Bean
    public Binding orderDlxBinding() {
        return BindingBuilder.bind(orderDlxQueue())
                .to(orderDlxExchange()).with(ORDER_DLX_ROUTING_KEY);
    }
}

package com.example.demo.mq;

import com.example.demo.entity.Order;
import com.example.demo.service.OrderService;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 延迟关单消费者：监听死信队列。
 * 消息 30 分钟后才到这里，必须重新查库判断订单是否已支付——这就是「兜底」。
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OrderCloseConsumer {

    private final OrderService orderService;

    @RabbitListener(queues = RabbitOrderConfig.ORDER_DLX_QUEUE)
    public void onOrderClose(OrderCloseMessage message, Channel channel,
                             @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        log.info("收到延迟关单消息，orderId={}", message.getOrderId());
        try {
            // 幂等兜底：可能用户刚好在 30 分钟内完成了支付，此时不应关单
            Order order = orderService.getById(message.getOrderId());
            if (order != null && "UNPAID".equals(order.getStatus())) {
                orderService.closeOrder(message.getOrderId());
                log.info("订单已超时关闭，orderId={}", message.getOrderId());
            } else {
                log.info("订单状态为 {}，无需关闭，orderId={}", order == null ? "NULL" : order.getStatus(),
                        message.getOrderId());
            }
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("关单处理异常，orderId={}", message.getOrderId(), e);
            // 关单失败不重回队列（会立即再次变死信循环），记录日志人工补偿
            channel.basicNack(deliveryTag, false, false);
        }
    }
}

// 下单服务里：创建订单成功后发两条消息
orderMessageProducer.sendOrderCreatedEvent(event, String.valueOf(orderId));       // 即时：扣库存/发通知
orderMessageProducer.sendOrderCloseMessage(new OrderCloseMessage(orderId), "close:" + orderId); // 延迟：30 分钟后关单

@RabbitListener(queues = RabbitOrderConfig.ORDER_CREATED_QUEUE,
        concurrency = "4")   // 4 个消费者线程并发处理
public void onOrderCreated(...) { ... }

spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 4        # 初始 4 个消费者线程
        max-concurrency: 10   # 高峰最多扩到 10 个

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      # acks=all：消息写入所有 ISR 副本才算成功，配合重试保证不丢
      acks: all
      retries: 3
      # 发送失败重试时，若开启幂等可避免重试产生重复（spring-kafka 默认开启 enable-idempotence）
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: order-group
      # 关闭自动提交，改由代码显式 ack，避免「还没处理完就提交」导致丢消息
      enable-auto-commit: false
      # 新消费组从最早消息开始读（生产可按需改 latest）
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        # 反序列化时信任的包，避免 ClassCastException
        spring.json.trusted.packages: "com.example.demo.*"
    listener:
      # manual_immediate：方法处理完成后立即手动提交 offset
      ack-mode: manual_immediate

package com.example.demo.mq;

import com.example.demo.entity.OrderCreatedEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/**
 * Kafka 订单事件生产者。
 * key = 订单号，保证同一订单的消息路由到同一分区（保序）。
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaOrderProducer {

    public static final String TOPIC_ORDER_EVENTS = "order-events";

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public void sendOrderEvent(OrderCreatedEvent event) {
        // 用 orderId 作 key：同一订单进同一分区，分区内严格有序
        String key = String.valueOf(event.getOrderId());
        CompletableFuture<SendResult<String, OrderCreatedEvent>> future =
                kafkaTemplate.send(TOPIC_ORDER_EVENTS, key, event);
        // 异步感知发送结果：失败记日志，配合 acks=all 保证不丢
        future.whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("Kafka 发送失败，orderId={}", event.getOrderId(), ex);
            } else {
                log.info("Kafka 发送成功，orderId={}, partition={}, offset={}",
                        event.getOrderId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            }
        });
    }
}

package com.example.demo.mq;

import com.example.demo.service.OrderEventService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * Kafka 订单事件消费者（生产级）。
 * 手动提交 offset + 幂等：失败不提交，重读后靠幂等保证不重复处理。
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class KafkaOrderConsumer {

    private final OrderEventService orderEventService;

    @KafkaListener(topics = KafkaOrderProducer.TOPIC_ORDER_EVENTS, groupId = "order-group")
    public void onOrderEvent(OrderCreatedEvent event, Acknowledgment ack) {
        log.info("Kafka 收到订单事件，orderId={}", event.getOrderId());
        try {
            // 处理内部必须幂等（唯一键 / Redis 去重）：因为失败重试会重复收到同一条消息
            orderEventService.process(event);
            // 处理成功才提交 offset，崩溃后从该处重读，不丢
            ack.acknowledge();
        } catch (Exception e) {
            log.error("Kafka 订单事件处理失败，orderId={}，不提交 offset，等待重试", event.getOrderId(), e);
            // 不调用 acknowledge()：offset 不前进，稍后重读该消息
            // 注意：生产上要设重试上限，坏消息进死信/告警，否则会无限重试
        }
    }
}

package com.example.demo.service.impl;

import com.example.demo.entity.OrderCreatedEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.time.Duration;

/**
 * 幂等处理的通用模式：Redis setNX 去重。
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class OrderEventServiceImpl implements OrderEventService {

    private final StringRedisTemplate redisTemplate;

    @Override
    public void process(OrderCreatedEvent event) {
        // 业务唯一标识：这里用「事件类型 + 订单号」，也可用消息自带的消息 id
        String idempotentKey = "idempotent:order:" + event.getOrderId();

        // setIfAbsent 返回 true 表示「第一次抢到」，返回 false 表示「已处理过，跳过」
        Boolean firstTime = redisTemplate.opsForValue()
                .setIfAbsent(idempotentKey, "1", Duration.ofHours(24));
        if (!Boolean.TRUE.equals(firstTime)) {
            log.warn("重复消息，已跳过，orderId={}", event.getOrderId());
            return; // 幂等：直接返回，不再重复处理
        }

        // 真正的业务处理（扣库存、加积分等）
        doBusiness(event);
    }

    private void doBusiness(OrderCreatedEvent event) {
        // 实际业务逻辑，这里仅示意
        log.info("处理订单事件成功，orderId={}", event.getOrderId());
    }
}

@Slf4j
@Service
public class NotifyService {

    @Async
    public void sendEmail(String to, String content) {
        // 模拟耗时操作：发邮件
        log.info("发邮件线程：{}", Thread.currentThread().getName());
    }
}

package com.example.demo.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * 异步任务配置：自定义线程池，禁止用默认 SimpleAsyncTaskExecutor。
 */
@Configuration
@EnableAsync  // 开启 @Async 支持（缺了这个注解 @Async 全部失效）
public class AsyncConfig {

    @Bean("asyncExecutor")
    public ThreadPoolTaskExecutor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 核心线程数：常驻线程，任务不多时也保留这些线程待命
        executor.setCorePoolSize(8);
        // 最大线程数：核心线程忙不过来、队列也满了时，最多扩到这个数
        executor.setMaxPoolSize(16);
        // 队列容量：核心线程都忙时，新任务先进入队列排队等待
        executor.setQueueCapacity(500);
        // 空闲线程存活时间：超过核心线程数的线程，空闲这么久后被回收
        executor.setKeepAliveSeconds(60);
        // 线程名前缀：方便在日志/堆栈里定位是哪类任务
        executor.setThreadNamePrefix("async-");
        // 拒绝策略：队列满了且线程也满了时，新任务交给调用线程同步执行（不丢任务）
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 等待任务都完成后才关闭线程池（配合 Spring 优雅停机）
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setAwaitTerminationSeconds(30);
        // 必须调用 initialize()，否则线程池不生效
        executor.initialize();
        return executor;
    }
}

package com.example.demo.service;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

/**
 * 通知服务：异步发短信、发邮件。
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class NotifyService {

    /** 显式指定线程池，避免落到默认 SimpleAsyncTaskExecutor */
    @Async("asyncExecutor")
    public void sendEmail(String to, String content) {
        log.info("开始发邮件，to={}, 线程={}", to, Thread.currentThread().getName());
        try {
            // 模拟发送邮件（生产上是调用邮件网关 SDK）
            Thread.sleep(2000);
            log.info("邮件发送成功，to={}", to);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 恢复中断标记，遵循线程中断规范
            log.error("邮件发送被中断，to={}", to);
        }
    }
}

@Service
@RequiredArgsConstructor
public class OrderServiceImpl implements OrderService {

    private final NotifyService notifyService;

    @Override
    public Long createOrder(OrderCreateDTO dto) {
        // ... 下单主流程（落库等）...
        Long orderId = 123L;

        // 异步发通知：主线程不等待，立即返回
        notifyService.sendEmail(dto.getEmail(), "您的订单 " + orderId + " 已创建");
        log.info("下单完成，orderId={}，通知已异步发出", orderId);
        return orderId;
    }
}

@Async("asyncExecutor")
public CompletableFuture<String> fetchUserName(Long userId) {
    String name = userMapper.selectById(userId).getName();
    return CompletableFuture.completedFuture(name);
}

// 调用方：可以先做别的事，最后统一取结果
CompletableFuture<String> f1 = userService.fetchUserName(1L);
CompletableFuture<String> f2 = userService.fetchUserName(2L);
String name1 = f1.get();  // 阻塞等待结果
String name2 = f2.get();

@Service
public class OrderService {

    @Async("asyncExecutor")
    public void asyncMethod() {
        log.info("线程：{}", Thread.currentThread().getName());
    }

    public void caller() {
        // 坑：this 是原始对象，不是代理，@Async 失效，这里会同步执行
        this.asyncMethod();
    }
}

@RestController
@RequiredArgsConstructor
public class AsyncTestController {

    private final OrderService orderService;

    @GetMapping("/async/test")
    public ApiResult<String> test() {
        // 1. 外部调用：走代理，@Async 生效，线程名是 async-1 之类
        orderService.asyncMethod();

        // 2. 自调用：不走代理，@Async 失效，线程名是 http-nio-8080-exec-1（主线程）
        orderService.caller();
        return ApiResult.ok("查看日志中的线程名即可验证");
    }
}

@Service
@RequiredArgsConstructor
public class OrderService {
    private final AsyncTaskService asyncTaskService; // 注入另一个 Bean

    public void caller() {
        asyncTaskService.asyncMethod(); // 跨 Bean 调用，走代理，@Async 生效
    }
}

@Service
public class AsyncTaskService {
    @Async("asyncExecutor")
    public void asyncMethod() {
        log.info("线程：{}", Thread.currentThread().getName());
    }
}

package com.example.demo.config;

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;

/** 开启定时任务支持 */
@Configuration
@EnableScheduling
public class ScheduleConfig {
}

package com.example.demo.task;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class DemoTask {

    /** cron：每天凌晨 2 点执行 */
    @Scheduled(cron = "0 0 2 * * ?")
    public void dailyReport() {
        log.info("生成每日报表");
    }

    /** fixedRate：每 60 秒执行（上次开始后 60s，不管上次是否结束） */
    @Scheduled(fixedRate = 60000)
    public void syncData() {
        log.info("同步数据");
    }

    /** fixedDelay：上次执行结束后延迟 5 秒再执行（不会重叠） */
    @Scheduled(fixedDelay = 5000)
    public void cleanTempFile() {
        log.info("清理临时文件");
    }
}

package com.example.demo.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;

import java.util.concurrent.Executors;

/**
 * 定时任务配置：自定义调度线程池，避免多任务互相阻塞。
 */
@Configuration
@EnableScheduling
public class ScheduleConfig implements SchedulingConfigurer {

    @Override
    public void configureTasks(ScheduledTaskRegistrar registrar) {
        // 核心线程数按「同时可能触发的定时任务数」定，这里给 4 个足够
        registrar.setScheduler(Executors.newScheduledThreadPool(4));
    }
}

package com.example.demo.task;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Collections;
import java.util.UUID;

/**
 * 分布式定时任务：用 Redis 分布式锁保证多实例下只执行一次。
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ReportTask {

    private final StringRedisTemplate redisTemplate;

    private static final String LOCK_KEY = "lock:dailyReport";
    // 释放锁的 Lua 脚本：先比对 value 再删除，防止误删别人的锁（value 相同才删）
    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT = new DefaultRedisScript<>(
            "if redis.call('get', KEYS[1]) == ARGV[1] then " +
            "  return redis.call('del', KEYS[1]) " +
            "else return 0 end", Long.class);

    @Scheduled(cron = "0 0 2 * * ?")
    public void dailyReport() {
        // 锁的唯一 value：用 UUID 区分「这把锁是谁加的」
        String lockValue = UUID.randomUUID().toString();

        // 抢锁：setIfAbsent 返回 true 表示抢到，false 表示其他实例已在执行
        Boolean locked = redisTemplate.opsForValue()
                .setIfAbsent(LOCK_KEY, lockValue, Duration.ofMinutes(30));

        if (!Boolean.TRUE.equals(locked)) {
            log.info("其他实例正在执行报表任务，本实例跳过");
            return; // 没抢到锁，直接跳过，避免重复执行
        }

        try {
            // 只有抢到锁的实例才真正执行任务
            log.info("开始生成每日报表...");
            doGenerateReport();
        } finally {
            // 安全释放：只有锁还是自己的才删（防止任务超时后锁过期、误删别人新加的锁）
            redisTemplate.execute(UNLOCK_SCRIPT, Collections.singletonList(LOCK_KEY), lockValue);
        }
    }

    private void doGenerateReport() {
        // 实际报表生成逻辑
        log.info("报表生成完成");
    }
}

CREATE TABLE `local_message` (
    `id`          BIGINT       NOT NULL AUTO_INCREMENT COMMENT '主键',
    `biz_type`    VARCHAR(64)  NOT NULL COMMENT '业务类型，如 ORDER_CREATED',
    `biz_key`     VARCHAR(128) NOT NULL COMMENT '业务唯一键，用于幂等和去重',
    `payload`     TEXT         NOT NULL COMMENT '消息体（JSON）',
    `status`      TINYINT      NOT NULL DEFAULT 0 COMMENT '0=待投递 1=已投递',
    `retry_count` INT          NOT NULL DEFAULT 0 COMMENT '已重试次数',
    `create_time` DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    `update_time` DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    PRIMARY KEY (`id`),
    UNIQUE KEY `uk_biz_key` (`biz_key`)  -- 唯一索引：保证同一业务不重复投递
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COMMENT = '本地消息表';

package com.example.demo.entity;

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

import java.time.LocalDateTime;

@Data
@TableName("local_message")
public class LocalMessage {

    @TableId(type = IdType.AUTO)
    private Long id;

    /** 业务类型 */
    private String bizType;

    /** 业务唯一键（如订单号），用于幂等 */
    private String bizKey;

    /** 消息体 JSON */
    private String payload;

    /** 0=待投递 1=已投递 */
    private Integer status;

    /** 重试次数 */
    private Integer retryCount;

    private LocalDateTime createTime;

    private LocalDateTime updateTime;
}

package com.example.demo.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.example.demo.entity.LocalMessage;
import org.apache.ibatis.annotations.Mapper;

@Mapper
public interface LocalMessageMapper extends BaseMapper<LocalMessage> {
}

package com.example.demo.service.impl;

import com.alibaba.fastjson2.JSON;
import com.example.demo.dto.OrderCreateDTO;
import com.example.demo.entity.LocalMessage;
import com.example.demo.entity.Order;
import com.example.demo.entity.OrderCreatedEvent;
import com.example.demo.mapper.LocalMessageMapper;
import com.example.demo.mapper.OrderMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/**
 * 下单服务：本地消息表模式。
 * 关键：业务数据 + 消息记录在【同一个本地事务】里写入。
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class OrderServiceImpl implements OrderService {

    private final OrderMapper orderMapper;
    private final LocalMessageMapper messageMapper;

    @Override
    @Transactional(rollbackFor = Exception.class) // 关键：同库同事务
    public Long createOrder(OrderCreateDTO dto) {
        // 1. 写业务数据
        Order order = new Order();
        order.setUserId(dto.getUserId());
        order.setAmount(dto.getAmount());
        order.setStatus("UNPAID");
        orderMapper.insert(order);
        Long orderId = order.getId();

        // 2. 写消息记录（与上面在同一个事务里，保证不丢）
        OrderCreatedEvent event = new OrderCreatedEvent(orderId, dto.getUserId(), dto.getAmount());
        LocalMessage message = new LocalMessage();
        message.setBizType("ORDER_CREATED");
        message.setBizKey(String.valueOf(orderId));   // 唯一键 = 订单号
        message.setPayload(JSON.toJSONString(event)); // 消息体 JSON
        message.setStatus(0);                          // 待投递
        message.setRetryCount(0);
        messageMapper.insert(message);

        // 3. 事务提交后，两条数据同时落库；之后由定时任务投递 MQ
        log.info("订单创建成功，orderId={}，消息记录已同事务写入", orderId);
        return orderId;
    }
}

package com.example.demo.task;

import com.example.demo.entity.LocalMessage;
import com.example.demo.mapper.LocalMessageMapper;
import com.example.demo.mq.OrderMessageProducer;
import com.alibaba.fastjson2.JSON;
import com.example.demo.entity.OrderCreatedEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 本地消息表投递任务：定时扫描待投递消息，发送到 MQ。
 * 这里同样要防多实例重复投递——靠消息表的唯一键 + 状态机兜底。
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class MessageRelayTask {

    private final LocalMessageMapper messageMapper;
    private final OrderMessageProducer orderMessageProducer;

    /** 每 5 秒扫描一次待投递消息 */
    @Scheduled(fixedDelay = 5000)
    public void relay() {
        // 只查待投递的消息，一次最多 100 条，避免一次扫太多
        List<LocalMessage> messages = messageMapper.selectList(
                new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<LocalMessage>()
                        .eq(LocalMessage::getStatus, 0)
                        .last("limit 100"));

        for (LocalMessage msg : messages) {
            try {
                OrderCreatedEvent event = JSON.parseObject(msg.getPayload(), OrderCreatedEvent.class);
                // 投递 MQ（RabbitMQ 的 confirm 机制已保证 Broker 收到）
                orderMessageProducer.sendOrderCreatedEvent(event, msg.getBizKey());
                // 投递成功，更新状态为已投递
                msg.setStatus(1);
                messageMapper.updateById(msg);
            } catch (Exception e) {
                log.error("消息投递失败，bizKey={}", msg.getBizKey(), e);
                // 失败不更新状态，下次扫描继续重试；可结合 retry_count 做告警
            }
        }
    }
}

-- 订单表（避免与 MySQL 关键字 order 冲突，用 orders）
CREATE TABLE `orders` (
    `id`          BIGINT        NOT NULL AUTO_INCREMENT COMMENT '订单号',
    `user_id`     BIGINT        NOT NULL COMMENT '用户 id',
    `amount`      DECIMAL(10,2) NOT NULL COMMENT '金额',
    `status`      VARCHAR(16)   NOT NULL DEFAULT 'UNPAID' COMMENT '状态：UNPAID/PAID/CLOSED',
    `create_time` DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP,
    `update_time` DATETIME      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    PRIMARY KEY (`id`)
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COMMENT = '订单表';

-- 本地消息表（第 6 章已给出建表语句，此处省略，直接复制第 6 章 6.3 的 SQL 即可）

package com.example.demo.entity;

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

import java.math.BigDecimal;
import java.time.LocalDateTime;

@Data
@TableName("orders")
public class Order {

    @TableId(type = IdType.AUTO)
    private Long id;

    private Long userId;

    private BigDecimal amount;

    /** UNPAID / PAID / CLOSED */
    private String status;

    private LocalDateTime createTime;

    private LocalDateTime updateTime;
}

package com.example.demo.dto;

import jakarta.validation.constraints.DecimalMin;
import jakarta.validation.constraints.NotNull;
import lombok.Data;

import java.math.BigDecimal;

@Data
public class OrderCreateDTO {

    @NotNull(message = "用户 id 不能为空")
    private Long userId;

    @NotNull(message = "金额不能为空")
    @DecimalMin(value = "0.01", message = "金额必须大于 0")
    private BigDecimal amount;
}

package com.example.demo.entity;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.math.BigDecimal;

/** 订单创建事件：消息体（JSON 序列化后存入消息表 / 投递 MQ） */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class OrderCreatedEvent {
    private Long orderId;
    private Long userId;
    private BigDecimal amount;
}