【系统架构设计百科】CQRS：读写分离的架构哲学

// 符合 CQS 的设计
public class Stack<T> {
    public void push(T item) { items.add(item); }          // Command
    public void pop() { items.remove(items.size() - 1); }  // Command
    public T top() { return items.get(items.size() - 1); }  // Query
    public int size() { return items.size(); }               // Query
}

// 写模型：包含业务规则
type Course struct {
    ID       string
    Title    string
    Price    int64
    Status   CourseStatus
    Chapters []Chapter
}

func (c *Course) Publish() error {
    if len(c.Chapters) == 0 {
        return errors.New("课程至少包含一个章节才能发布")
    }
    c.Status = CourseStatusPublished
    return nil
}

// 读模型：为列表展示优化
type CourseListItem struct {
    ID           string
    Title        string
    Price        int64
    ChapterCount int
    StudentCount int
    Rating       float64
}

-- 写侧主表：标准化设计
CREATE TABLE courses (
    id            BIGINT PRIMARY KEY,
    title         VARCHAR(200) NOT NULL,
    description   TEXT,
    price         BIGINT NOT NULL,
    status        VARCHAR(20) NOT NULL DEFAULT 'draft',
    instructor_id BIGINT NOT NULL REFERENCES instructors(id),
    created_at    TIMESTAMP NOT NULL DEFAULT NOW(),
    updated_at    TIMESTAMP NOT NULL DEFAULT NOW()
);

-- 读侧物化视图：反标准化设计
CREATE MATERIALIZED VIEW course_list_view AS
SELECT c.id, c.title, c.price, c.status,
       i.name AS instructor_name,
       COUNT(DISTINCT ch.id) AS chapter_count,
       COUNT(DISTINCT e.id) AS enroll_count,
       COALESCE(AVG(r.score), 0) AS avg_rating
FROM courses c
JOIN instructors i ON c.instructor_id = i.id
LEFT JOIN chapters ch ON c.id = ch.course_id
LEFT JOIN enrollments e ON c.id = e.course_id
LEFT JOIN reviews r ON c.id = r.course_id
WHERE c.status = 'published'
GROUP BY c.id, c.title, c.price, c.status, i.name;

// 写侧：标准 CRUD
func (s *CourseCommandService) UpdateCourse(ctx context.Context, cmd UpdateCourseCommand) error {
    course, err := s.repo.FindByID(ctx, cmd.CourseID)
    if err != nil {
        return fmt.Errorf("查询课程失败: %w", err)
    }
    course.Title = cmd.Title
    course.Price = cmd.Price
    course.UpdatedAt = time.Now()

    if err := s.repo.Save(ctx, course); err != nil {
        return fmt.Errorf("保存课程失败: %w", err)
    }
    s.sync.OnCourseUpdated(ctx, course)
    return nil
}

// 读侧：专门的查询服务
func (s *CourseQueryService) ListCourses(ctx context.Context, q CourseListQuery) ([]CourseListItem, error) {
    return s.readRepo.Search(ctx, SearchParams{
        Keyword: q.Keyword, Category: q.Category,
        SortBy: q.SortBy, Offset: q.Offset, Limit: q.Limit,
    })
}

func (s *CourseCommandService) UpdateCourse(ctx context.Context, cmd UpdateCourseCommand) error {
    // ... 写操作 ...
    return s.eventBus.Publish(ctx, "course.updated", CourseUpdatedEvent{
        CourseID: course.ID, Title: course.Title, Price: course.Price,
    })
}

# Debezium CDC 连接器配置
name: course-connector
config:
  connector.class: io.debezium.connector.postgresql.PostgresConnector
  database.hostname: write-db
  database.port: 5432
  database.dbname: courses
  table.include.list: public.courses,public.chapters
  topic.prefix: coursedb
  plugin.name: pgoutput

func (s *CourseService) PublishCourse(ctx context.Context, courseID string) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // 更新写模型
    _, err = tx.ExecContext(ctx,
        "UPDATE courses SET status = 'published', updated_at = NOW() WHERE id = $1", courseID)
    if err != nil {
        return fmt.Errorf("更新课程状态失败: %w", err)
    }

    // 同一事务中更新读模型
    _, err = tx.ExecContext(ctx, `
        INSERT INTO course_search_index (id, title, price, status)
        SELECT c.id, c.title, c.price, 'published' FROM courses c WHERE c.id = $1
        ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, updated_at = NOW()
    `, courseID)
    if err != nil {
        return fmt.Errorf("更新搜索索引失败: %w", err)
    }

    return tx.Commit()
}

// 读侧消费者
func (h *CourseProjectionHandler) HandleCoursePublished(ctx context.Context, evt CoursePublishedEvent) error {
    course, err := h.writeRepo.FindByID(ctx, evt.CourseID)
    if err != nil {
        return fmt.Errorf("读取课程数据失败: %w", err)
    }
    return h.readStore.Upsert(ctx, buildCourseReadModel(course))
}

func (r *Rebuilder) Rebuild(ctx context.Context) error {
    newIndex := fmt.Sprintf("courses_%d", time.Now().Unix())
    if err := r.readStore.CreateIndex(ctx, newIndex); err != nil {
        return err
    }

    var lastID int64
    for {
        rows, _ := r.writeDB.QueryContext(ctx,
            "SELECT * FROM courses WHERE id > $1 ORDER BY id LIMIT $2", lastID, 500)
        batch := scanAll(rows)
        if len(batch) == 0 {
            break
        }
        r.readStore.BulkInsert(ctx, newIndex, batch)
        lastID = batch[len(batch)-1].ID
    }

    return r.readStore.SwitchAlias(ctx, "courses", newIndex)
}

func (s *CourseQueryService) GetCourse(ctx context.Context, courseID string, opts QueryOptions) (*CourseDetail, error) {
    if opts.AfterVersion > 0 {
        readModel, _ := s.readRepo.FindByID(ctx, courseID)
        if readModel != nil && readModel.Version >= opts.AfterVersion {
            return readModel, nil
        }
        return s.writeRepo.FindByIDAsReadModel(ctx, courseID) // 回退到写库
    }
    return s.readRepo.FindByID(ctx, courseID)
}

function useCourseUpdate() {
  const queryClient = useQueryClient();

  return useMutation({
    mutationFn: (data: UpdateCourseData) => api.updateCourse(data),
    onMutate: async (newData) => {
      await queryClient.cancelQueries({ queryKey: ['course', newData.id] });
      const previous = queryClient.getQueryData(['course', newData.id]);
      queryClient.setQueryData(['course', newData.id], (old: Course) => ({
        ...old, ...newData, updatedAt: new Date().toISOString(),
      }));
      return { previous };
    },
    onError: (_err, newData, context) => {
      if (context?.previous) {
        queryClient.setQueryData(['course', newData.id], context.previous);
      }
    },
    onSettled: (_data, _error, variables) => {
      queryClient.invalidateQueries({ queryKey: ['course', variables.id] });
    },
  });
}

async function updateAndWaitForSync(courseId: string, data: UpdateCourseData): Promise<Course> {
  const result = await api.updateCourse(courseId, data);
  const targetVersion = result.version;

  for (let i = 0; i < 10; i++) {
    const course = await api.getCourse(courseId);
    if (course.version >= targetVersion) return course;
    await sleep(300);
  }
  return api.getCourseFromWriteSide(courseId);
}

func (r *ReadRouter) GetCourse(ctx context.Context, userID, courseID string) (*CourseDetail, error) {
    key := fmt.Sprintf("%s:%s", userID, courseID)
    if writeTime, ok := r.writeCache.Get(key); ok {
        if time.Since(writeTime) < r.ttl {
            return r.getFromWriteDB(ctx, courseID) // 写后短窗口内走写库
        }
    }
    return r.readService.GetCourse(ctx, courseID) // 正常走读库
}

package command

type CreateCourseCommand struct {
    Title        string
    Description  string
    Price        int64
    InstructorID string
}

type CreateCourseHandler struct {
    repo      CourseRepository
    publisher EventPublisher
}

func (h *CreateCourseHandler) Handle(ctx context.Context, cmd CreateCourseCommand) error {
    if cmd.Title == "" {
        return fmt.Errorf("课程标题不能为空")
    }
    course := &Course{
        ID: generateID(), Title: cmd.Title, Description: cmd.Description,
        Price: cmd.Price, InstructorID: cmd.InstructorID,
        Status: StatusDraft, CreatedAt: time.Now(), Version: 1,
    }
    if err := h.repo.Save(ctx, course); err != nil {
        return fmt.Errorf("保存课程失败: %w", err)
    }
    return h.publisher.Publish(ctx, CourseCreatedEvent{
        CourseID: course.ID, Title: course.Title, CreatedAt: course.CreatedAt,
    })
}

package query

type ListCoursesQuery struct {
    Keyword  string
    SortBy   string
    Page     int
    PageSize int
}

type ListCoursesHandler struct {
    readStore CourseReadStore
}

func (h *ListCoursesHandler) Handle(ctx context.Context, q ListCoursesQuery) (*CourseListResult, error) {
    if q.PageSize <= 0 || q.PageSize > 100 {
        q.PageSize = 20
    }
    return h.readStore.SearchCourses(ctx, q)
}

func (a *CourseAPI) HandleCreateCourse(w http.ResponseWriter, r *http.Request) {
    var cmd command.CreateCourseCommand
    if err := json.NewDecoder(r.Body).Decode(&cmd); err != nil {
        http.Error(w, "请求格式错误", http.StatusBadRequest)
        return
    }
    if err := a.createHandler.Handle(r.Context(), cmd); err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    w.WriteHeader(http.StatusCreated)
}

func (a *CourseAPI) HandleListCourses(w http.ResponseWriter, r *http.Request) {
    q := query.ListCoursesQuery{
        Keyword:  r.URL.Query().Get("keyword"),
        Page:     parseIntOrDefault(r.URL.Query().Get("page"), 1),
        PageSize: parseIntOrDefault(r.URL.Query().Get("page_size"), 20),
    }
    result, err := a.listHandler.Handle(r.Context(), q)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    json.NewEncoder(w).Encode(result)
}

// 命令与处理
public record CreateCourseCommand(String title, String description, long price, String instructorId) {}

@Service
public class CourseCommandHandler {
    private final CourseRepository courseRepository;
    private final ApplicationEventPublisher eventPublisher;

    @Transactional
    public String handle(CreateCourseCommand cmd) {
        if (cmd.price() < 0) throw new IllegalArgumentException("课程价格不能为负数");
        Course course = Course.create(cmd.title(), cmd.description(), cmd.price(), cmd.instructorId());
        courseRepository.save(course);
        eventPublisher.publishEvent(new CourseCreatedEvent(course.getId(), course.getTitle(), Instant.now()));
        return course.getId();
    }
}

// 查询服务（直接 SQL，不经过领域模型）
@Service
public class CourseQueryService {
    private final JdbcTemplate jdbcTemplate;

    public List<CourseListItemDTO> listPublished(String keyword, int page, int size) {
        String sql = """
            SELECT c.id, c.title, c.price, i.name AS instructor_name,
                   COUNT(DISTINCT ch.id) AS chapter_count,
                   COALESCE(AVG(r.score), 0) AS avg_rating
            FROM courses c
            JOIN instructors i ON c.instructor_id = i.id
            LEFT JOIN chapters ch ON c.id = ch.course_id
            LEFT JOIN reviews r ON c.id = r.course_id
            WHERE c.status = 'published' AND c.title ILIKE ?
            GROUP BY c.id, c.title, c.price, i.name
            ORDER BY chapter_count DESC LIMIT ? OFFSET ?
            """;
        return jdbcTemplate.query(sql,
            new Object[]{"%" + keyword + "%", size, (page - 1) * size},
            (rs, n) -> new CourseListItemDTO(rs.getString("id"), rs.getString("title"),
                rs.getLong("price"), rs.getString("instructor_name"),
                rs.getInt("chapter_count"), rs.getDouble("avg_rating")));
    }
}

// 异步投影
@Component
public class CourseReadModelProjection {
    @EventListener
    @Async
    public void on(CourseCreatedEvent event) {
        // 从写库读取完整数据，构建读模型写入读库
    }
}