在Java中實現(xiàn)CQRS架構(gòu)的全過程
更新時間：2026年03月04日 14:10:20 作者：中華人民共和國程序員
CQRS是Command?Query?Responsibility?Segregation的縮寫,一般稱作命令查詢職責分離,本文給大家介紹在Java中實現(xiàn)CQRS架構(gòu)的全過程,感興趣的朋友跟隨小編一起看看吧
使用中介者模式輕松實現(xiàn)命令查詢職責分離，構(gòu)建高內(nèi)聚、低耦合的應用系統(tǒng)
一、知識點回顧
1. 什么是CQRS？
CQRS是Command Query Responsibility Segregation的縮寫，一般稱作命令查詢職責分離。從字面意思理解，就是將命令（寫入）和查詢（讀取）的責任劃分到不同的模型中。
對比一下常用的 CRUD 模式（創(chuàng)建-讀取-更新-刪除），通常我們會讓用戶界面與負責所有四種操作的數(shù)據(jù)存儲交互。而 CQRS 則將這些操作分成兩種模式，一種用于查詢（又稱 "R"），另一種用于命令（又稱 "CUD"）。
2. CQRS的作用是什么？
CQRS將系統(tǒng)的寫操作（命令）和讀操作（查詢）分離到不同的模型和數(shù)據(jù)存儲中，從而實現(xiàn)讀寫分離，提高系統(tǒng)的性能、可擴展性和安全性，并使復雜業(yè)務邏輯（寫端）和高效查詢（讀端）各自得到優(yōu)化，降低系統(tǒng)復雜性。它允許為寫操作設計嚴謹?shù)念I(lǐng)域模型，為讀操作設計簡單、只關(guān)注查詢效率的數(shù)據(jù)模型（如專用視圖或報表數(shù)據(jù)庫），并可通過事件等機制保持最終一致性。
3. CQRS 的優(yōu)點
- 獨立縮放。 CQRS 使讀取模型和寫入模型能夠獨立縮放。 此方法可幫助最大程度地減少鎖爭用并提高負載下的系統(tǒng)性能。
- 優(yōu)化的數(shù)據(jù)架構(gòu)。 讀取操作可以使用針對查詢進行優(yōu)化的模式。 寫入操作使用針對更新優(yōu)化的模式。
- 安全性。 通過分隔讀取和寫入，可以確保只有適當?shù)挠驅(qū)嶓w或操作有權(quán)對數(shù)據(jù)執(zhí)行寫入操作。
- 關(guān)注點分離。 分離讀取和寫入責任會導致更簡潔、更易于維護的模型。 寫入端通常處理復雜的業(yè)務邏輯。 讀取端可以保持簡單且專注于查詢效率。
- 更簡單的查詢。 在讀取數(shù)據(jù)庫中存儲具體化視圖時，應用程序可以在查詢時避免復雜的聯(lián)接。
二、關(guān)于PipelinR
項目地址
項目開發(fā)者在Github的介紹不多，關(guān)鍵是最后一句話：It's similar to a popular MediatR .NET library. 意思就是這個項目是參考著一個叫MediatR的.net庫寫的。關(guān)于MediatR我之前有兩篇文章專門介紹過。
PipelinR（包括MediatR）提供了一種CQRS的實現(xiàn)方式，基于中介者模式實現(xiàn)進程內(nèi)消息傳遞，用于解耦應用中的各個組件，支持請求/響應（一對一，有返回值）和發(fā)布/訂閱（一對多，無返回值）兩種消息模式。它們在內(nèi)部提供管道行為 (Pipeline Behaviors)，用于在消息處理前后插入自定義邏輯，如日志、驗證、異常處理等。
需要提醒的是，PipelinR并不是一個完整的CQRS框架，它只是一個中介者模式的具體實現(xiàn)方式，將調(diào)用方和處理方進行了解耦，而這種模式恰好可以用來在一個單體應用（或者是微服務的服務內(nèi)部）中實現(xiàn)簡單的CQRS。
三、依賴安裝和配置
1. Maven安裝
<dependency> <groupId>net.sizovs</groupId> <artifactId>pipelinr</artifactId> <version>0.11</version> </dependency>
2. Gradle安裝
dependencies {
compile 'net.sizovs:pipelinr:0.11'
}在Spring項目中配置PipelinR
@Configuration
public class PipelinrConfiguration {
@Bean
Pipeline pipeline(ObjectProvider<Command.Handler> commandHandlers, ObjectProvider<Notification.Handler> notificationHandlers, ObjectProvider<Command.Middleware> middlewares) {
return new Pipelinr()
.with(commandHandlers::stream)
.with(notificationHandlers::stream)
.with(middlewares::orderedStream);
}
}四、核心組件
Pipeline/Pipelinr：Pipeline是消息和處理器之間的中介者，調(diào)用方向Pipeline發(fā)送消息，Pipeline收到消息后通過注冊到Pipeline的中間件進行層層傳遞并最終抵達匹配的消息處理器進行處理。Pipelinr是Pipeline的默認實現(xiàn)。Command<R>：用于約定請求/響應模式的消息類型，泛型參數(shù)R是返回值的類型，如果不需要返回值，可以將R指定為Voidy。Notification：用于約定發(fā)布/訂閱模式的消息類型，沒有返回值，消息可以有多個處理器。Middleware：管道中間件，Command和Notification都定義了各自的中間件接口。Pipeline接收到的消息，在到達最終的處理器之前，會經(jīng)過所有注冊到Pipeline的中間?？梢允褂肕iddleware實現(xiàn)諸如日志記錄、數(shù)據(jù)驗證、開啟事務等一系列操作。
五、請求/響應模式實現(xiàn)
請求/響應模式需要用到Command接口。
1. 定義Command
Command代表一個請求，需要實現(xiàn)net.sizovs.pipelinr.Command接口。泛型參數(shù)指定返回值類型。
// 定義一個創(chuàng)建用戶的命令
public class CreateUserCommand implements Command<UserResponse> {
private String username;
private String email;
public CreateUserCommand(String username, String email) {
this.username = username;
this.email = email;
}
public String getUsername() {
return username;
}
public String getEmail() {
return email;
}
}
// 返回值類型
public class UserResponse {
private Long userId;
private String username;
private String email;
public UserResponse(Long userId, String username, String email) {
this.userId = userId;
this.username = username;
this.email = email;
}
// getters
}2. 定義Command Handler
創(chuàng)建該Command對應的處理器，實現(xiàn)net.sizovs.pipelinr.Command.Handler接口。
@Component
public class CreateUserCommandHandler implements Command.Handler<CreateUserCommand, UserResponse> {
@Autowired
private UserRepository userRepository;
@Override
public UserResponse handle(CreateUserCommand command) {
// 業(yè)務邏輯處理
User user = new User();
user.setUsername(command.getUsername());
user.setEmail(command.getEmail());
User savedUser = userRepository.save(user);
return new UserResponse(savedUser.getId(), savedUser.getUsername(), savedUser.getEmail());
}
}3. 在業(yè)務代碼中使用
通過注入Pipeline實例，發(fā)送Command并獲取響應。
@Service
public class UserService {
@Autowired
private Pipeline pipeline;
public UserResponse createUser(String username, String email) {
CreateUserCommand command = new CreateUserCommand(username, email);
UserResponse response = pipeline.send(command);
return response;
}
}4. 添加Command中間件
中間件可以在Command處理前后執(zhí)行一些操作，如驗證、日志、事務管理等。
@Component
public class LoggingMiddleware implements Command.Middleware {
private static final Logger logger = LoggerFactory.getLogger(LoggingMiddleware.class);
@Override
public <R, C extends Command<R>> R invoke(C command, Chain<R> chain) {
logger.info("Executing command: {}", command.getClass().getSimpleName());
try {
R result = chain.proceed(command);
logger.info("Command executed successfully");
return result;
} catch (Exception e) {
logger.error("Command execution failed", e);
throw e;
}
}
}
@Component
public class ValidationMiddleware implements Command.Middleware {
@Autowired
private Validator validator;
@Override
public <R, C extends Command<R>> R invoke(C command, Chain<R> chain) {
Set<ConstraintViolation<C>> violations = validator.validate(command);
if (!violations.isEmpty()) {
throw new ConstraintViolationException("Validation failed", violations);
}
return chain.proceed(command);
}
}
@Component
@Order(1) // 指定中間件執(zhí)行順序
public class TransactionMiddleware implements Command.Middleware {
@Autowired
private PlatformTransactionManager transactionManager;
@Override
public <R, C extends Command<R>> R invoke(C command, Chain<R> chain) {
TransactionStatus status = transactionManager.getTransaction(new DefaultTransactionDefinition());
try {
R result = chain.proceed(command);
transactionManager.commit(status);
return result;
} catch (Exception e) {
transactionManager.rollback(status);
throw e;
}
}
}六、發(fā)布/訂閱模式實現(xiàn)
發(fā)布/訂閱模式使用Notification接口，用于一對多的消息分發(fā)，沒有返回值。
1. 定義Notification
Notification代表一個事件通知，需要實現(xiàn)net.sizovs.pipelinr.Notification接口。
// 定義一個用戶創(chuàng)建成功的事件通知
public class UserCreatedNotification implements Notification {
private Long userId;
private String username;
private String email;
private LocalDateTime createdTime;
public UserCreatedNotification(Long userId, String username, String email) {
this.userId = userId;
this.username = username;
this.email = email;
this.createdTime = LocalDateTime.now();
}
// getters
}2. 定義Notification Handler
Notification可以有多個處理器，每個處理器實現(xiàn)net.sizovs.pipelinr.Notification.Handler接口。
@Component
public class SendWelcomeEmailHandler implements Notification.Handler<UserCreatedNotification> {
private static final Logger logger = LoggerFactory.getLogger(SendWelcomeEmailHandler.class);
@Autowired
private EmailService emailService;
@Override
public void handle(UserCreatedNotification notification) {
logger.info("Sending welcome email to user: {}", notification.getUsername());
emailService.sendWelcomeEmail(notification.getEmail(), notification.getUsername());
}
}
@Component
public class LogUserCreationHandler implements Notification.Handler<UserCreatedNotification> {
private static final Logger logger = LoggerFactory.getLogger(LogUserCreationHandler.class);
@Autowired
private UserAuditLogRepository auditLogRepository;
@Override
public void handle(UserCreatedNotification notification) {
logger.info("Logging user creation: {}", notification.getUsername());
UserAuditLog auditLog = new UserAuditLog();
auditLog.setUserId(notification.getUserId());
auditLog.setOperation("CREATE");
auditLog.setTimestamp(notification.getCreatedTime());
auditLogRepository.save(auditLog);
}
}
@Component
public class UpdateUserStatisticsHandler implements Notification.Handler<UserCreatedNotification> {
private static final Logger logger = LoggerFactory.getLogger(UpdateUserStatisticsHandler.class);
@Autowired
private UserStatisticsRepository statisticsRepository;
@Override
public void handle(UserCreatedNotification notification) {
logger.info("Updating statistics for new user: {}", notification.getUsername());
UserStatistics stats = statisticsRepository.findOrCreate();
stats.incrementTotalUsers();
statisticsRepository.save(stats);
}
}3. 發(fā)送Notification
在Command處理完成后，可以發(fā)送Notification通知所有相關(guān)的處理器。
@Component
public class CreateUserCommandHandler implements Command.Handler<CreateUserCommand, UserResponse> {
@Autowired
private UserRepository userRepository;
@Autowired
private Pipeline pipeline;
@Override
public UserResponse handle(CreateUserCommand command) {
// 業(yè)務邏輯處理
User user = new User();
user.setUsername(command.getUsername());
user.setEmail(command.getEmail());
User savedUser = userRepository.save(user);
// 發(fā)送事件通知
UserCreatedNotification notification = new UserCreatedNotification(
savedUser.getId(),
savedUser.getUsername(),
savedUser.getEmail()
);
pipeline.send(notification);
return new UserResponse(savedUser.getId(), savedUser.getUsername(), savedUser.getEmail());
}
}4. 添加Notification中間件
類似Command，Notification也支持中間件。
@Component
public class NotificationLoggingMiddleware implements Notification.Middleware {
private static final Logger logger = LoggerFactory.getLogger(NotificationLoggingMiddleware.class);
@Override
public <N extends Notification> void invoke(N notification, Chain chain) {
logger.info("Publishing notification: {}", notification.getClass().getSimpleName());
try {
chain.proceed(notification);
logger.info("Notification published successfully");
} catch (Exception e) {
logger.error("Notification publishing failed", e);
throw e;
}
}
}
@Component
public class NotificationErrorHandlingMiddleware implements Notification.Middleware {
private static final Logger logger = LoggerFactory.getLogger(NotificationErrorHandlingMiddleware.class);
@Override
public <N extends Notification> void invoke(N notification, Chain chain) {
try {
chain.proceed(notification);
} catch (Exception e) {
logger.error("Error handling notification: {}", notification.getClass().getSimpleName(), e);
// 可以選擇吞掉異?；蛑匦聮伋觯Q于業(yè)務需求
// throw e;
}
}
}七、總結(jié)
核心收獲
通過本文的介紹，我們了解了如何在Java應用中使用PipelinR框架實現(xiàn)CQRS模式。核心要點總結(jié)如下：
1. CQRS的價值
- 讀寫分離：通過Command處理寫操作，Notification處理事件響應，實現(xiàn)職責的明確劃分
- 獨立優(yōu)化：讀端和寫端可以獨立優(yōu)化，不同的數(shù)據(jù)模型適應不同的場景需求
- 系統(tǒng)解耦：中介者模式解耦了調(diào)用方和處理方，提高了系統(tǒng)的可維護性和可擴展性
2. PipelinR的核心特性
- 輕量級實現(xiàn)：相比完整的CQRS框架，PipelinR更輕便，學習成本低
- 靈活的管道機制：通過中間件可以方便地植入橫切關(guān)注點（如日志、驗證、事務等）
- 支持兩種消息模式：Command用于請求/響應，Notification用于發(fā)布/訂閱
3. 最佳實踐建議
- 合理使用中間件：通過@Order注解控制中間件執(zhí)行順序，但要避免中間件層級過多導致性能問題
- 異常處理：根據(jù)場景選擇合適的異常處理策略，Notification可考慮不中斷其他處理器的錯誤隔離
- 事件驅(qū)動設計：充分利用Notification實現(xiàn)事件驅(qū)動架構(gòu)，解耦不同的業(yè)務流程
- 代碼組織：按照Command、Handler、Middleware的劃分方式組織代碼，保持結(jié)構(gòu)清晰
實施建議
適用場景
- 中等復雜度的業(yè)務系統(tǒng)，需要良好的代碼結(jié)構(gòu)和可維護性
- 業(yè)務邏輯相對復雜，需要事件驅(qū)動的系統(tǒng)設計
- 團隊具備良好的DDD設計理念和架構(gòu)意識
注意事項
- 學習曲線：雖然PipelinR本身簡單，但要理解CQRS的設計理念需要一定時間
- 適度使用：CQRS不是銀彈，過度設計會增加系統(tǒng)復雜度，要根據(jù)實際需求決定是否引入
- 團隊協(xié)作：CQRS的有效實施對團隊的整體架構(gòu)意識和編碼規(guī)范要求較高
- 性能考慮：雖然使用了中介者模式會引入少量額外開銷，但對大多數(shù)應用來說可以忽略不計
結(jié)論
PipelinR提供了一種輕量級、簡潔的CQRS實現(xiàn)方案。它特別適合那些想要在不過度復雜化系統(tǒng)的前提下，引入DDD思想和事件驅(qū)動設計的項目。通過合理運用Command和Notification，結(jié)合恰當?shù)闹虚g件設計，開發(fā)者可以構(gòu)建出高內(nèi)聚、低耦合、易于維護和擴展的應用系統(tǒng)。
關(guān)鍵是要把握好"度"——既要充分發(fā)揮CQRS和PipelinR的優(yōu)勢，又要避免為了追求"高大上"的架構(gòu)而過度設計，最終的目標是為業(yè)務的快速迭代和長期維護提供支撐。
到此這篇關(guān)于在Java中實現(xiàn)CQRS架構(gòu)的全過程的文章就介紹到這了,更多相關(guān)java cqrs架構(gòu)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
AOP面向切面編程，它是一種思想，它是對某一類事情的集中處理，而AOP是一種思想，而Spring AOP是一個框架，提供了一種對AOP思想的實現(xiàn)，它們的關(guān)系和loC與DI類似，這篇文章主要介紹了Spring AOP統(tǒng)一功能處理示例代碼,需要的朋友可以參考下2023-01-01
Spring?框架中的?Bean?作用域（Scope）使用詳解
Spring框架中的Bean作用域（Scope）決定了在應用程序中創(chuàng)建和管理的Bean對象的生命周期和可見性。本文將詳細介紹Spring框架中的Bean作用域的不同類型，包括Singleton、Prototype、Request、Session和Application，并解釋它們的特點和適用場景。2023-09-09
這篇文章主要介紹了Java通過URL類下載圖片,文中結(jié)合實例代碼補充介紹了java通過url獲取圖片文件的相關(guān)知識，代碼簡單易懂，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-02-02
HashMap在不同的JDK版本下的實現(xiàn)是不同的，在JDK1.7時，HashMap 底層是通過數(shù)組+鏈表實現(xiàn)的；而在JDK1.8時，HashMap底層是通過數(shù)組+鏈表或紅黑樹實現(xiàn)的，本詳細介紹了HashMap底層是如何實現(xiàn)的,需要的朋友可以參考下2023-05-05
springboot?項目啟動后無日志輸出直接結(jié)束的解決
這篇文章主要介紹了springboot?項目啟動后無日志輸出直接結(jié)束的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
這篇文章主要幫助大家輕松掌握Java狀態(tài)模式,具有一定的參考價值，感興趣的小伙伴們可以參考一下2016-09-09
IDEA2020.1構(gòu)建Spring5.2.x源碼的方法
這篇文章主要介紹了IDEA2020.1構(gòu)建Spring5.2.x源碼的方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-10-10
本文主要介紹了SpringBoot項目更換項目名稱,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2024-06-06
使用javaweb項目對數(shù)據(jù)庫增、刪、改、查操作的實現(xiàn)方法
這篇文章主要給大家介紹了關(guān)于使用javaweb項目對數(shù)據(jù)庫增、刪、改、查操作的實現(xiàn)方法,avaWeb是指使用Java語言進行Web應用程序開發(fā)的技術(shù),可以利用Java編寫一些動態(tài)網(wǎng)頁、交互式網(wǎng)頁、企業(yè)級應用程序等,需要的朋友可以參考下2023-07-07
Java Socket編程簡介_動力節(jié)點Java學院整理
這篇文章主要介紹了Java Socket編程簡介的相關(guān)知識，非常不錯，具有參考借鑒價值,需要的朋友可以參考下2017-05-05