事件驱动：ApplicationEvent 与消息队列集成

@Servicepublic class OrderService {    private final OrderRepository orderRepository;    private final SmsService smsService;    private final PointService pointService;    private final CouponService couponService;    private final AuditLogService auditLogService;    private final WarehouseService warehouseService;    public OrderService(OrderRepository orderRepository, SmsService smsService,                        PointService pointService, CouponService couponService,                        AuditLogService auditLogService, WarehouseService warehouseService) {        this.orderRepository = orderRepository;        this.smsService = smsService;        this.pointService = pointService;        this.couponService = couponService;        this.auditLogService = auditLogService;        this.warehouseService = warehouseService;    }    @Transactional    public Order placeOrder(PlaceOrderCommand cmd) {        Order order = orderRepository.save(Order.create(cmd));        // 核心动作 —— 存订单        // 下面全是"顺带要做的事"，却和核心动作挤在一起        smsService.send(cmd.userId(), "下单成功");        pointService.add(cmd.userId(), order.getAmount());        couponService.issue(cmd.userId(), order.getId());        auditLogService.record("PLACE_ORDER", order.getId());        warehouseService.notifyPrepare(order.getId());        return order;    }}

package com.example.order.event;import java.math.BigDecimal;import java.time.LocalDateTime;// 一个普通的不可变对象即可作为事件，无需继承任何基类public record OrderPlacedEvent(Long orderId, Long userId, BigDecimal amount, LocalDateTime placedAt) {}

@Transactionalpublic Order placeOrder(PlaceOrderCommand cmd) {    Order order = orderRepository.save(Order.create(cmd));    // 只表达"订单已创建"这个事实，不关心谁处理、怎么处理    publisher.publishEvent(new OrderPlacedEvent(            order.getId(), cmd.userId(), order.getAmount(), LocalDateTime.now()));    return order;}

@RestControllerpublic class DemoController {    private static final Logger log = LoggerFactory.getLogger(DemoController.class);    private final ApplicationEventPublisher publisher;    public DemoController(ApplicationEventPublisher publisher) {        this.publisher = publisher;    }    @GetMapping("/place")    public String place() {        log.info("发布事件，当前线程 = {}", Thread.currentThread().getName());        publisher.publishEvent(new OrderPlacedEvent(1L, 100L, new BigDecimal("9.9"), LocalDateTime.now()));        log.info("publishEvent 返回，当前线程 = {}", Thread.currentThread().getName());        return "ok";    }}

@Componentpublic class SmsListener {    private static final Logger log = LoggerFactory.getLogger(SmsListener.class);    @EventListener    public void onOrderPlaced(OrderPlacedEvent event) {        log.info("监听器执行，当前线程 = {}", Thread.currentThread().getName());    }}

发布事件，当前线程 = http-nio-8080-exec-1监听器执行，当前线程 = http-nio-8080-exec-1publishEvent 返回，当前线程 = http-nio-8080-exec-1

@Configuration@EnableAsyncpublic class EventAsyncConfig implements AsyncConfigurer {    private static final Logger log = LoggerFactory.getLogger(EventAsyncConfig.class);    @Bean("eventExecutor")    public Executor eventExecutor() {        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();        executor.setCorePoolSize(4);        executor.setMaxPoolSize(16);        executor.setQueueCapacity(200);        executor.setThreadNamePrefix("evt-");   // 便于在日志里识别        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());        executor.initialize();        return executor;    }    // 异步方法抛出的异常不会传播到调用方，必须在这里兜底    @Override    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {        return (ex, method, params) -> log.error("异步事件执行失败: {}", method.getName(), ex);    }}

spring:  rabbitmq:    host: localhost    port: 5672    username: guest    password: guest    publisher-confirm-type: correlated   # 开启生产者确认    publisher-returns: true              # 不可路由的消息回调    listener:      simple:        acknowledge-mode: manual         # 手动 ACK，消费成功才确认        retry:          enabled: true          max-attempts: 3                # 本地重试 3 次

@Configurationpublic class RabbitConfig {    @Bean    public TopicExchange orderExchange() {        return ExchangeBuilder.topicExchange("order.exchange").durable(true).build();    }    @Bean    public Queue pointQueue() {        return QueueBuilder.durable("order.point.queue").build();    }    @Bean    public Binding pointBinding(Queue pointQueue, TopicExchange orderExchange) {        // order.placed.xxx 的消息都会进入积分队列        return BindingBuilder.bind(pointQueue).to(orderExchange).with("order.placed.#");    }}

@Componentpublic class OrderMessagePublisher {    private final RabbitTemplate rabbitTemplate;    public OrderMessagePublisher(RabbitTemplate rabbitTemplate) {        this.rabbitTemplate = rabbitTemplate;    }    public void publish(OrderPlacedEvent event) {        rabbitTemplate.convertAndSend("order.exchange", "order.placed", event);    }}

CREATE TABLE t_local_message (  id              BIGINT       PRIMARY KEY AUTO_INCREMENT,  biz_type        VARCHAR(64)  NOT NULL COMMENT '业务类型，如 ORDER_PLACED',  biz_id          BIGINT       NOT NULL COMMENT '业务主键，用于幂等',  payload         JSON         NOT NULL COMMENT '消息体',  status          TINYINT      NOT NULL DEFAULT 0 COMMENT '0待发送 1已发送 2已消费',  retry_count     INT          NOT NULL DEFAULT 0 COMMENT '重试次数',  next_retry_time DATETIME     NOT NULL COMMENT '下次重试时间',  create_time     DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,  UNIQUE KEY uk_biz (biz_type, biz_id)   -- 唯一键天然保证幂等);

@Transactionalpublic Order placeOrder(PlaceOrderCommand cmd) {    Order order = orderRepository.save(Order.create(cmd));    // 与订单在同一个事务里写入本地消息，保证"数据在则消息在"    localMessageRepository.save(LocalMessage.pending(            "ORDER_PLACED", order.getId(), toPayload(order)));    return order;}

@Componentpublic class LocalMessageCompensator {    private static final Logger log = LoggerFactory.getLogger(LocalMessageCompensator.class);    private final LocalMessageRepository repository;    private final RabbitTemplate rabbitTemplate;    public LocalMessageCompensator(LocalMessageRepository repository, RabbitTemplate rabbitTemplate) {        this.repository = repository;        this.rabbitTemplate = rabbitTemplate;    }    @Scheduled(fixedDelay = 5000)    public void compensate() {        for (LocalMessage message : repository.findDue(100)) {            try {                rabbitTemplate.convertAndSend("order.exchange", message.bizType(), message.payload());                repository.markSent(message.getId());            } catch (Exception ex) {                log.error("消息补偿发送失败: id={}", message.getId(), ex);                repository.bumpRetry(message.getId(), nextRetryTime());            }        }    }}

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">    <modelVersion>4.0.0</modelVersion>    <parent>        <groupId>org.springframework.boot</groupId>        <artifactId>spring-boot-starter-parent</artifactId>        <version>3.3.4</version>        <relativePath/>    </parent>    <groupId>com.example</groupId>    <artifactId>event-lab</artifactId>    <version>0.0.1-SNAPSHOT</version>    <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-jdbc</artifactId>        </dependency>        <dependency>            <groupId>com.h2database</groupId>            <artifactId>h2</artifactId>            <scope>runtime</scope>        </dependency>    </dependencies>    <build>        <plugins>            <plugin>                <groupId>org.springframework.boot</groupId>                <artifactId>spring-boot-maven-plugin</artifactId>            </plugin>        </plugins>    </build></project>

spring:  datasource:    url: jdbc:h2:mem:lab;DB_CLOSE_DELAY=-1    driver-class-name: org.h2.Driver  sql:    init:      mode: always      schema-locations: classpath:schema.sqllogging:  level:    com.example.eventlab: DEBUG  pattern:    console: "%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n"

CREATE TABLE IF NOT EXISTS t_order (  id      BIGINT PRIMARY KEY AUTO_INCREMENT,  user_id BIGINT       NOT NULL,  amount  DECIMAL(10,2) NOT NULL);

package com.example.eventlab;import java.math.BigDecimal;public record OrderPlacedEvent(Long orderId, Long userId, BigDecimal amount) {}

package com.example.eventlab;import java.util.concurrent.ThreadLocalRandom;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.context.event.EventListener;import org.springframework.scheduling.annotation.Async;import org.springframework.stereotype.Component;import org.springframework.transaction.event.TransactionPhase;import org.springframework.transaction.event.TransactionalEventListener;@Componentpublic class LabListeners {    private static final Logger log = LoggerFactory.getLogger(LabListeners.class);    @EventListener    public void syncListen(OrderPlacedEvent e) {        log.info("[同步] orderId={} thread={}", e.orderId(), Thread.currentThread().getName());    }    @Async("eventExecutor")    @EventListener    public void asyncListen(OrderPlacedEvent e) throws InterruptedException {        long cost = ThreadLocalRandom.current().nextInt(300, 800);        Thread.sleep(cost);        log.info("[异步] orderId={} thread={} cost={}ms", e.orderId(), Thread.currentThread().getName(), cost);    }    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)    public void afterCommit(OrderPlacedEvent e) {        log.info("[AFTER_COMMIT] orderId={} thread={}", e.orderId(), Thread.currentThread().getName());    }    @TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)    public void afterRollback(OrderPlacedEvent e) {        log.warn("[AFTER_ROLLBACK] orderId={} 本次作废", e.orderId());    }}

package com.example.eventlab;import java.math.BigDecimal;import java.util.Map;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Value;import org.springframework.context.ApplicationEventPublisher;import org.springframework.jdbc.core.JdbcTemplate;import org.springframework.scheduling.annotation.EnableAsync;import org.springframework.scheduling.annotation.Async;import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.web.bind.annotation.*;import org.springframework.transaction.annotation.Transactional;@RestController@EnableAsyncclass EventLabController {    private static final Logger log = LoggerFactory.getLogger(EventLabController.class);    private final JdbcTemplate jdbc;    private final ApplicationEventPublisher publisher;    EventLabController(JdbcTemplate jdbc, ApplicationEventPublisher publisher) {        this.jdbc = jdbc;        this.publisher = publisher;    }    @Bean("eventExecutor")    ThreadPoolTaskExecutor eventExecutor() {        ThreadPoolTaskExecutor ex = new ThreadPoolTaskExecutor();        ex.setCorePoolSize(4);        ex.setMaxPoolSize(8);        ex.setQueueCapacity(100);        ex.setThreadNamePrefix("evt-");        return ex;    }    @PostMapping("/orders")    @Transactional    Map<String, Object> create(@RequestParam(defaultValue = "false") boolean boom) {        jdbc.update("INSERT INTO t_order(user_id, amount) VALUES (?, ?)", 100L, new BigDecimal("9.90"));        Long id = jdbc.queryForObject("SELECT MAX(id) FROM t_order", Long.class);        log.info("准备发布事件 thread={}", Thread.currentThread().getName());        publisher.publishEvent(new OrderPlacedEvent(id, 100L, new BigDecimal("9.90")));        if (boom) {            throw new IllegalStateException("故意回滚");        }        return Map.of("id", id, "status", "created");    }}

# 正常下单：同步监听器与 AFTER_COMMIT 都会出现curl -s -X POST "http://localhost:8080/orders"# 故意回滚：同步监听器照样跑了（坑！），AFTER_COMMIT 没有，改走 AFTER_ROLLBACKcurl -s -X POST "http://localhost:8080/orders?boom=true"

{"id":1,"status":"created"}

14:32:05.112 [http-nio-8080-exec-1] INFO  c.e.e.EventLabController  - 准备发布事件 thread=http-nio-8080-exec-114:32:05.113 [http-nio-8080-exec-1] INFO  c.e.e.LabListeners        - [同步] orderId=1 thread=http-nio-8080-exec-114:32:05.114 [http-nio-8080-exec-1] INFO  c.e.e.EventLabController  - ... （控制器返回）14:32:05.118 [http-nio-8080-exec-1] INFO  c.e.e.LabListeners        - [AFTER_COMMIT] orderId=1 thread=http-nio-8080-exec-114:32:05.621 [evt-2]                 INFO  c.e.e.LabListeners        - [异步] orderId=1 thread=evt-2 cost=509ms