笑话人生

年华易逝 懂得珍惜

0%

Spring Boot 事件驱动编程：ApplicationEventPublisher 原理与实践指南

发表于 分类于 Disqus：
本文字数： 12k 阅读时长 ≈ 11 分钟

基本介绍

ApplicationEventPublisher 是 Spring 框架的核心接口，用于发布应用事件，实现观察者模式。其核心作用包括：

  1. 事件发布：允许组件发布自定义事件
  2. 松耦合：实现发布者与订阅者的解耦
  3. 同步处理：默认同步执行（可通过 @Async 实现异步）
  4. 继承机制：事件对象可继承扩展（支持 ApplicationEvent 或任意 POJO）

工作流程：

1
[发布者] → (发布事件) → [ApplicationContext] → (路由事件) → [监听器]

应用场景

  1. 业务解耦：如用户注册后发送邮件/短信
  2. 状态变更通知：订单状态变化时更新库存
  3. 审计日志：关键操作后记录审计信息
  4. 异步任务触发：耗时操作异步执行
  5. 系统监控：关键事件触发监控上报

代码示例

添加 Maven 依赖

由于 Spring Boot 已经内置了事件发布机制，我们只需要引入 spring-boot-starter 即可，它包含了 spring-context，其中就有 ApplicationEventPublisher。

1
2
3
4
5
6
7
8
9
10
11
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

关于 Spring Boot 入门可以参考：Spring Boot 入门指南：从零开始创建 Web 应用

事件定义（POJO）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import lombok.Getter;
import org.springframework.context.ApplicationEvent;

// 这里用了 Lombok 注解，也可以参考上一篇文章
@Getter
public class UserRegisterEvent extends ApplicationEvent {

    private final String username;

    public UserRegisterEvent(Object source, String username) {
        // source 通常是事件发布者
        super(source);
        this.username = username;
    }

}

事件监听器

1
2
3
4
5
6
7
8
9
10
@Component
public class UserRegisterListener {

    // 监听方式1：注解监听指定事件，此时 UserRegisterEvent 无需继承 ApplicationEvent
    @EventListener
    public void handleEvent(UserRegisterEvent event) {
        System.out.println("[注解监听] 新用户注册: " + event.getUsername());
    }

}
1
2
3
4
5
6
7
8
9
// 监听方式2：实现 ApplicationListener 接口，此时 UserRegisterEvent 需要继承 ApplicationEvent
@Component
public class EmailListener implements ApplicationListener<UserRegisterEvent> {

    @Override
    public void onApplicationEvent(UserRegisterEvent event) {
        System.out.println("[接口监听] 发送欢迎邮件至: " + event.getUsername());
    }
}

在 Spring 4.2 之前，自定义事件必须继承 ApplicationEvent。从 Spring 4.2 开始，事件可以是任意对象，不再强制要求继承 ApplicationEvent。因此，有两种解决方案：

  1. 让 UserRegisterEvent 继承 ApplicationEvent（这样两种监听方式都支持）
  2. 将实现 ApplicationListener 接口的监听器改为使用 @EventListener 注解（推荐，因为更灵活）

为了保持代码的简洁和现代 Spring 的使用方式，我们通常推荐使用 @EventListener 注解。这里为了演示两种方式，我们让事件类继承 ApplicationEvent，同时，在发布事件的时候，需要传递 source（通常就是发布者对象，但也可以为 null）

事件发布服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@Service
public class UserService {
    
    // 注入事件发布器
    private final ApplicationEventPublisher publisher;

    public UserService(ApplicationEventPublisher publisher) {
        this.publisher = publisher;
    }

    public void registerUser(String username) {
        System.out.println("注册用户: " + username);

        // 发布事件，需传递事件源（通常就是发布者对象 this，但也可以为 null）和业务数据
        publisher.publishEvent(new UserRegisterEvent(this, username));

        System.out.println("主流程完成");
    }
}

主应用类

1
2
3
4
5
6
7
8
9
10
@SpringBootApplication
public class ApplicationEventPublisherDemoApplication {

	public static void main(String[] args) {
		ConfigurableApplicationContext context = SpringApplication.run(ApplicationEventPublisherDemoApplication.class, args);
		UserService userService = context.getBean(UserService.class);
		userService.registerUser("cylong");
	}

}

代码解释

  1. 事件对象：UserRegisterEvent 封装事件数据
  2. 监听器：
    • 注解方式：@EventListener 自动匹配事件类型（也就是根据类名匹配）
    • 接口方式：实现 ApplicationListener 接口
  3. 发布器：
    • ApplicationEventPublisher.publishEvent() 触发事件
    • Spring 自动注入发布器实例
  4. 执行流程：
    • 主应用调用 registerUser()
    • 服务内部发布事件
    • 所有监听器同步执行

运行输出

1
2
3
4
注册用户: cylong
[接口监听] 发送欢迎邮件至: cylong
[注解监听] 新用户注册: cylong
主流程完成

异常处理

下面将展示三种异常处理方式（局部捕获、全局处理和异步处理）的实现

代码示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import org.springframework.context.ApplicationEvent;

// 1. 自定义事件
public class OrderEvent extends ApplicationEvent {

    private final String orderId;

    public OrderEvent(Object source, String orderId) {
        super(source);
        this.orderId = orderId;
    }

    public String getOrderId() {
        return orderId;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

@Component
public class OrderEventListeners {

    // 2.1 局部异常捕获 - 库存服务
    @EventListener
    public void handleInventory(OrderEvent event) {
        try {
            System.out.println("\n[库存服务] 处理订单: " + event.getOrderId());

            if ("error_stock".equals(event.getOrderId())) {
                throw new RuntimeException("库存不足，商品A缺货");
            }

            System.out.println("   > 库存锁定成功");
        } catch (Exception e) {
            System.out.println("   ! [库存异常] " + e.getMessage());
            System.out.println("   > 执行本地补偿: 释放预留资源");
        }
    }

    // 2.2 局部异常捕获 - 支付服务
    @EventListener
    public void handlePayment(OrderEvent event) {
        try {
            System.out.println("\n[支付服务] 处理订单: " + event.getOrderId());

            if ("error_payment".equals(event.getOrderId())) {
                throw new RuntimeException("支付失败，信用卡余额不足");
            }

            System.out.println("   > 支付处理成功");
        } catch (Exception e) {
            System.out.println("   ! [支付异常] " + e.getMessage());
            System.out.println("   > 执行本地补偿: 取消支付预授权");
        }
    }

    // 3. 用于演示全局异常处理的监听器（不捕获异常）
    @EventListener
    public void handleNotification(OrderEvent event) {
        System.out.println("\n[通知服务] 处理订单: " + event.getOrderId());

        if ("error_stock".equals(event.getOrderId())) {
            throw new RuntimeException("通知服务失败: 短信配额不足");
        }

        System.out.println("   > 已发送订单确认通知");
    }

    // 4. 异步异常处理
    @Async
    @EventListener
    public void handleAsyncTask(OrderEvent event) {
        System.out.println("\n[异步服务] 开始处理: " + event.getOrderId());

        if ("async_error".equals(event.getOrderId())) {
            throw new RuntimeException("异步任务处理失败: 外部API超时");
        }

        System.out.println("   > 异步任务完成");
    }
}
1
2
3
4
5
6
7
8
9
10
11
import org.springframework.util.ErrorHandler;

// 5. 同步事件全局异常处理器
public class SyncEventErrorHandler implements ErrorHandler {

    @Override
    public void handleError(Throwable t) {
        System.out.println("\n[全局同步异常处理器] 捕获异常: " + t.getMessage());
        System.out.println("   > 执行全局处理: 记录错误日志并告警");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;

import java.lang.reflect.Method;

// 6. 异步事件全局异常处理器
public class AsyncEventErrorHandler implements AsyncUncaughtExceptionHandler {

    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        System.out.println("\n[全局异步异常处理器] 捕获异常: " + ex.getMessage());
        System.out.println("   > 方法: " + method.getName());
        System.out.println("   > 参数类型: " + params[0].getClass().getSimpleName());
        System.out.println("   > 执行补偿操作: 记录日志并通知管理员");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;

@Service
public class OrderService {

    @Autowired
    private ApplicationEventPublisher eventPublisher;

    public void placeOrder(String orderId) {
        System.out.println("\n[主流程] 开始处理订单: " + orderId);
        try {
            // 模拟核心业务逻辑
            System.out.println("   > 创建订单记录");

            // 发布订单事件
            eventPublisher.publishEvent(new OrderEvent(this, orderId));

            System.out.println("[主流程] 订单处理完成: " + orderId);
        } catch (Exception e) {
            System.out.println("[主流程异常] " + e.getMessage());
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ApplicationEventMulticaster;
import org.springframework.context.event.SimpleApplicationEventMulticaster;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;

@Configuration
@EnableAsync
public class EventConfig implements AsyncConfigurer {

    // 配置异步事件广播器
    @Bean(name = "applicationEventMulticaster")
    public ApplicationEventMulticaster applicationEventMulticaster() {
        SimpleApplicationEventMulticaster eventMulticaster = new SimpleApplicationEventMulticaster();

        // 设置同步事件的异常处理器
        eventMulticaster.setErrorHandler(new SyncEventErrorHandler());

        // 设置异步执行器
        eventMulticaster.setTaskExecutor(new SimpleAsyncTaskExecutor());
        return eventMulticaster;
    }

    // 异步异常处理配置
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new AsyncEventErrorHandler();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
package com.cylong.applicationeventpublisherdemo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;

@SpringBootApplication
@EnableAsync
public class EventExceptionDemoApplication implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(EventExceptionDemoApplication.class, args);
    }

    @Autowired
    private OrderService orderService;

    @Override
    public void run(String... args) throws InterruptedException {
        System.out.println("\n===== 测试正常订单 =====");
        orderService.placeOrder("order_2023001");

        Thread.sleep(500); // 等待异步任务完成

        System.out.println("\n===== 测试库存不足订单 =====");
        orderService.placeOrder("error_stock");

        Thread.sleep(500);

        System.out.println("\n===== 测试支付失败订单 =====");
        orderService.placeOrder("error_payment");

        Thread.sleep(500);

        System.out.println("\n===== 测试异步异常订单 =====");
        orderService.placeOrder("async_error");

        Thread.sleep(1000); // 确保异步任务完成
    }

}

输出结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
===== 测试正常订单 =====

[主流程] 开始处理订单: order_2023001
   > 创建订单记录
[主流程] 订单处理完成: order_2023001

[库存服务] 处理订单: order_2023001

[支付服务] 处理订单: order_2023001

[通知服务] 处理订单: order_2023001
   > 已发送订单确认通知
   > 库存锁定成功
   > 支付处理成功

[异步服务] 开始处理: order_2023001
   > 异步任务完成

===== 测试库存不足订单 =====

[主流程] 开始处理订单: error_stock
   > 创建订单记录

[库存服务] 处理订单: error_stock

[支付服务] 处理订单: error_stock
   > 支付处理成功
   ! [库存异常] 库存不足，商品A缺货
   > 执行本地补偿: 释放预留资源
[主流程] 订单处理完成: error_stock

[通知服务] 处理订单: error_stock

[异步服务] 开始处理: error_stock

[全局同步异常处理器] 捕获异常: 通知服务失败: 短信配额不足
   > 执行全局处理: 记录错误日志并告警
   > 异步任务完成

===== 测试支付失败订单 =====

[主流程] 开始处理订单: error_payment
   > 创建订单记录

[库存服务] 处理订单: error_payment
   > 库存锁定成功

[支付服务] 处理订单: error_payment
[主流程] 订单处理完成: error_payment
   ! [支付异常] 支付失败，信用卡余额不足
   > 执行本地补偿: 取消支付预授权

[通知服务] 处理订单: error_payment
   > 已发送订单确认通知

[异步服务] 开始处理: error_payment
   > 异步任务完成

===== 测试异步异常订单 =====

[主流程] 开始处理订单: async_error
   > 创建订单记录

[库存服务] 处理订单: async_error
   > 库存锁定成功

[支付服务] 处理订单: async_error
   > 支付处理成功
[主流程] 订单处理完成: async_error

[通知服务] 处理订单: async_error
   > 已发送订单确认通知

[异步服务] 开始处理: async_error

[全局异步异常处理器] 捕获异常: 异步任务处理失败: 外部API超时
   > 方法: handleAsyncTask
   > 参数类型: OrderEvent
   > 执行补偿操作: 记录日志并通知管理员

异常处理机制解析

  1. 局部异常捕获：
    • 在监听器内部使用 try-catch 块
    • 示例：库存和支付服务的监听器
    • 特点：异常不会传播，不影响其他监听器
    • 适用场景：需要独立处理的业务异常
  2. 全局同步异常处理：
    • 实现 ErrorHandler 接口
    • 通过 SimpleApplicationEventMulticaster.setErrorHandler() 注册
    • 捕获所有未处理的同步事件异常
    • 示例：SyncEventErrorHandler
    • 特点：集中处理未被捕获的同步异常
  3. 全局异步异常处理：
    • 实现 AsyncUncaughtExceptionHandler 接口
    • 通过 AsyncConfigurer 配置
    • 捕获所有未处理的异步事件异常
    • 示例：AsyncEventErrorHandler
    • 特点：专用于异步执行场景

三种处理方式适用场景

处理方式 适用场景 优点 缺点
局部捕获 需要独立处理的业务异常 精确控制，不影响其他监听器 代码重复可能性高
全局同步处理 未捕获的同步异常统一处理 集中管理，避免异常传播 无法获取完整上下文
全局异步处理 所有未捕获的异步异常 统一处理异步任务失败 无法访问原始方法参数

注意事项

作用域限制

1
2
3
// 监听器需是 Spring 管理的 Bean
@Component
public class UserRegisterListener {...}

事件源（source）的作用

1
2
3
4
// 可获取事件发布者信息
if (event.getSource() instanceof UserService) {
    // 特殊处理
}

监听器执行顺序

监听器按注册顺序执行（可通过 @Order 调整）

1
2
3
@Order(1)  // 数字越小优先级越高
@EventListener
public void firstListener(UserRegisterEvent event) {...}

异步处理

1
2
3
4
5
@Async  // 启用异步
@EventListener
public void asyncHandle(UserRegisterEvent event) {
    // 耗时操作
}

需在配置类添加 @EnableAsync

1
2
3
4
5
6
7
8
9
10
11
@SpringBootApplication
@EnableAsync
public class ApplicationEventPublisherDemoApplication {

	public static void main(String[] args) {
		ConfigurableApplicationContext context = SpringApplication.run(ApplicationEventPublisherDemoApplication.class, args);
		UserService userService = context.getBean(UserService.class);
		userService.registerUser("cylong");
	}

}

事件继承

1
2
// 监听父类事件会同时接收子类事件
public class VIPRegisterEvent extends UserRegisterEvent {...}
  1. 监听 UserRegisterEvent 会接收到所有子类事件
  2. 使用 @EventListener(classes = VIPRegisterEvent.class) 限定具体类型
1
2
3
4
@EventListener(classes = VIPRegisterEvent.class)
public void handleEvent(UserRegisterEvent event) {
    System.out.println("[注解监听] 新 VIP 用户注册: " + event.getUsername());
}

性能建议

  1. 避免在监听器执行耗时操作（默认同步，可以切换为异步）
  2. 单个事件避免注册过多监听器

错误处理

  1. 监听器异常会传播到发布者
  2. 需要时添加单独异常处理

通过 ApplicationEventPublisher 可实现优雅的业务解耦，但需根据场景权衡同步/异步机制。在实际项目中，建议将核心业务与辅助操作（邮件、日志等）通过事件分离，提升系统可维护性。

分享精彩，留下足迹

欢迎关注我的其它发布渠道

Disqus 评论 LiveRe 评论