EDA - Spring Boot构建基于事件驱动的消息系统

文章目录

概述
事件驱动架构的基本概念
工程结构
Code
- 创建事件和事件处理器
- 创建事件总线
- 创建消息通道和发送逻辑
- 创建事件处理器
- 消息持久化
- 创建消息发送事件
- 配置 Spring Boot 启动类
- 测试
- 消息消费
- 运行项目

概述

在微服务架构和大规模分布式系统中，事件驱动架构（EDA）成为了非常重要的设计模式。通过事件驱动，我们可以解耦系统的各个组件，提高系统的可扩展性、可维护性和响应能力。

接下来，我们将演示一下如何在 Spring Boot 中实现一个基于事件驱动的消息发送和接收流程，从消息的发送、事件的发布到事件的监听。

事件驱动架构的基本概念

在事件驱动架构中，系统的各个组件通过事件进行通信。每个事件代表一个特定的行为或状态变化，当事件发布时，系统的其他部分可以响应这些事件并做出相应的处理。消息发送和接收的流程正是通过发布和监听事件来实现的。

接下来我们使用 Spring Boot 来实现一个基于事件驱动的消息系统。、

系统包含以下几个部分：

消息发送：消息将通过一个 MessageEventProcessor 进行处理，并且在处理完成后会发布一个事件。
事件发布：消息成功发送后，通过 ApplicationEventPublisher 发布一个 MessageSentEvent。
事件监听：一个监听器会接收到发布的事件并进行相应的处理（比如记录日志、通知其他组件等）

工程结构

在这里插入图片描述

EventBus：事件总线，负责发布事件。
MessageEventProcessor：处理消息事件的处理器。
Event、MessageEvent、MessageSentEvent：事件类，MessageEvent和MessageSentEvent继承自Event。
MessageChannel：消息通道接口，EmailMessageChannel是其具体实现。
MessageRepository：消息存储库，用于保存消息事件。
MessageChannelConfig：消息通道配置，配置了消息通道的Bean。
MessageController：消息控制器，处理发送消息的请求。
MessageSentEventListener：监听消息发送事件的监听器。

Code

创建事件和事件处理器

Event.java - 定义基础事件

package com.artisan.booteventbus.domain;public abstract class Event {// 事件的基本字段
}

MessageEvent.java - 定义具体的消息事件

package com.artisan.booteventbus.domain;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;import java.util.Map;@EqualsAndHashCode(callSuper = true)
@Data
@Slf4j
@NoArgsConstructor
@AllArgsConstructor
public class MessageEvent extends Event {private String message;private String channel;private Map<String, Object> metadata;}

EventHandler.java - 定义事件处理器接口

package com.artisan.booteventbus.bus;import com.artisan.booteventbus.domain.Event;public interface EventHandler<T extends Event> {void handle(T event);
}

创建事件总线

EventBus.java - 用于发布事件

package com.artisan.booteventbus.bus;import com.artisan.booteventbus.domain.Event;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;@Component
public class EventBus {private final ApplicationEventPublisher publisher;public EventBus(ApplicationEventPublisher publisher) {this.publisher = publisher;}public void publish(Event event) {publisher.publishEvent(event);}
}

创建消息通道和发送逻辑

MessageChannel.java - 定义消息通道接口

package com.artisan.booteventbus.service;import com.artisan.booteventbus.domain.MessageEvent;import java.util.concurrent.CompletableFuture;public interface MessageChannel {boolean supports(MessageEvent event);CompletableFuture<Void> sendAsync(MessageEvent event);
}

MessageChannelConfig.java - 初始化channel

package com.artisan.booteventbus.config;import com.artisan.booteventbus.service.MessageChannel;
import com.artisan.booteventbus.service.impl.EmailMessageChannel;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.ArrayList;
import java.util.List;@Configuration
public class MessageChannelConfig {@Beanpublic List<MessageChannel> messageChannels() {List<MessageChannel> channels = new ArrayList<>();channels.add(new EmailMessageChannel());// 可以继续添加其他类型的通道return channels;}
}

EmailMessageChannel.java - 实现邮件发送通道

package com.artisan.booteventbus.service.impl;import com.artisan.booteventbus.domain.MessageEvent;
import com.artisan.booteventbus.service.MessageChannel;
import lombok.extern.slf4j.Slf4j;import java.util.concurrent.CompletableFuture;@Slf4j
public class EmailMessageChannel implements MessageChannel {@Overridepublic boolean supports(MessageEvent event) {return "email".equals(event.getChannel());}@Overridepublic CompletableFuture<Void> sendAsync(MessageEvent event) {return CompletableFuture.runAsync(() -> {// 模拟邮件发送System.out.println(Thread.currentThread().getName() + "- Sending email: " + event.getMessage());log.info("Sending email: {}", event.getMessage());});}
}

创建事件处理器

MessageEventProcessor.java - 处理消息事件，保存事件并发送

package com.artisan.booteventbus.bus;import com.artisan.booteventbus.dao.MessageRepository;
import com.artisan.booteventbus.domain.MessageEvent;
import com.artisan.booteventbus.domain.MessageSentEvent;import com.artisan.booteventbus.service.MessageChannel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.scheduling.annotation.Async;import java.util.List;@Component
public class MessageEventProcessor implements EventHandler<MessageEvent> {private final EventBus eventBus;private final MessageRepository messageRepository;private final List<MessageChannel> channels;@Autowiredpublic MessageEventProcessor(EventBus eventBus, MessageRepository messageRepository, List<MessageChannel> channels) {this.eventBus = eventBus;this.messageRepository = messageRepository;this.channels = channels;}/*** @param event* Asyn 请使用自定义线程池，这里仅仅是 为了演示异步*/@Async@Overridepublic void handle(MessageEvent event) {// 1. 消息持久化messageRepository.save(event);// 2. 通道路由MessageChannel channel = channels.stream().filter(ch -> ch.supports(event)).findFirst().orElseThrow();// 3. 异步发送channel.sendAsync(event).thenRun(() -> eventBus.publish(new MessageSentEvent(event)));}}

消息持久化

MessageRepository.java - 用于消息的持久化（可以使用内存或数据库）

package com.artisan.booteventbus.dao;import com.artisan.booteventbus.domain.MessageEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Repository;import java.util.ArrayList;
import java.util.List;@Slf4j
@Repository
public class MessageRepository {private final List<MessageEvent> messageStore = new ArrayList<>();public void save(MessageEvent event) {// 模拟存储messageStore.add(event);System.out.println(Thread.currentThread().getName() + " - Message saved: " + event.getMessage());log.info("Message saved {}", event.getMessage());}
}

创建消息发送事件

MessageSentEvent.java - 定义发送后的事件

package com.artisan.booteventbus.domain;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;@EqualsAndHashCode(callSuper = true)
@Data
@NoArgsConstructor
@AllArgsConstructor
public class MessageSentEvent extends Event {private MessageEvent originalEvent;}

配置 Spring Boot 启动类

package com.artisan;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;@EnableAsync(proxyTargetClass=true)
@SpringBootApplication
public class BootEventBusApplication {public static void main(String[] args) {SpringApplication.run(BootEventBusApplication.class, args);}}

测试

为了测试整个架构，创建一个控制器来模拟发送消息。

package com.artisan.booteventbus.controller;import com.artisan.booteventbus.domain.MessageEvent;
import com.artisan.booteventbus.bus.EventBus;
import com.artisan.booteventbus.bus.MessageEventProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;import java.util.HashMap;@RestController
@RequestMapping("/messages")
public class MessageController {private final EventBus eventBus;private final MessageEventProcessor eventProcessor;@Autowiredpublic MessageController(EventBus eventBus, MessageEventProcessor eventProcessor) {this.eventBus = eventBus;this.eventProcessor = eventProcessor;}@RequestMapping("/send")public String sendMessage(@RequestParam String message, @RequestParam String channel) {MessageEvent event = new MessageEvent(message, channel, new HashMap<>());eventProcessor.handle(event); // 异步处理消息return "Message is being processed";}
}

消息消费

package com.artisan.booteventbus.listeners;import com.artisan.booteventbus.domain.MessageSentEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class MessageSentEventListener {@Async@EventListenerpublic void handleMessageSentEvent(MessageSentEvent event) {// 模拟处理事件System.out.println(Thread.currentThread().getName() + " - Received MessageSentEvent: " + event.getOriginalEvent().getMessage());log.info("Sending email: {}", event.getOriginalEvent().getMessage());}
}