arch-event-driven

arch-event-driven

用途

此技能使用 Kafka、RabbitMQ 以及事件溯源、CQRS、发布/订阅、死信队列和架构注册表等相关模式来实现事件驱动架构。它有助于在微服务环境中设计可扩展、低耦合的系统，以实时处理事件。

适用场景

当场景需要异步通信时，可使用此技能，例如微服务需要在不建立直接依赖的情况下响应事件。可将其用于高流量数据管道、实时分析，或必须解耦生产者与消费者的场景，例如电子商务订单处理或 IoT 数据流。对于轮询即可满足需求的简单同步操作，请避免使用此技能。

主要功能

• 为事件流设置 Kafka 主题和分区。
• 通过将事件存储在 Kafka 中来实现事件溯源，以便重建状态。
• 应用 CQRS，将读模型和写模型分离，并使用 Kafka 处理命令和查询。
• 在扇出场景中使用 Kafka 消费者组管理发布/订阅。
• 通过 Kafka 或 RabbitMQ 中的死信队列处理故障。
• 使用 Confluent Schema Registry 对 Avro 架构进行验证。

使用模式

若要实现发布/订阅，请创建 Kafka 主题，让生产者发布事件；消费者通过组订阅。若要实现事件溯源，请将所有状态变更存储为 Kafka 流中的事件，并重放这些事件以构建当前状态。在 CQRS 中，将命令路由到写入服务（例如通过 Kafka 生产者），将查询路由到读取服务（例如从物化视图读取）。通过配置 Kafka 主题将失败的消息重定向到死信队列。始终使用 JSON 或 Avro 格式定义事件架构，以确保一致性。

常用命令/API

使用 Kafka CLI 管理主题：运行 kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 --replication-factor 2。若要生成事件，请使用：

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders
{"orderId": 123, "status": "placed"}

若要消费事件，请使用：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --from-beginning

对于 RabbitMQ，可通过 CLI 声明交换机和队列：rabbitmqadmin declare exchange name=events type=fanout。API 端点：使用 Kafka REST Proxy 的 /topics/{topic}/messages，通过 POST 生成消息（例如 curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" --data '{"records":[{"value":{"orderId":123}}]}' http://localhost:8082/topics/orders）。通过请求头中的环境变量进行身份验证：`$KAFKA_API_KEY`，例如 -H "Authorization: Bearer $KAFKA_API_KEY"。配置格式：在生产者配置中使用 Kafka 属性文件，例如 key.serializer=org.apache.kafka.common.serialization.StringSerializer。

集成说明

将 Kafka 客户端库添加到应用程序中以集成 Kafka（例如，在 Java 中使用 kafka-clients:3.0.0）。设置凭据环境变量：export RABBITMQ_URL=amqp://user:$RABBITMQ_PASSWORD@localhost。对于架构注册表，在生产者配置中将其指向 Confluent 端点：schema.registry.url=http://localhost:8081。与数据库连接时，使用 Kafka Connect 配置 JDBC 源：通过 JSON 文件进行配置，例如 {"name": "jdbc-source", "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "connection.url": "jdbc:postgresql://localhost:5432/db"}。确保生产者在遇到暂时性错误时会重试。

错误处理

在 Kafka 中设置单独的故障主题以配置死信队列：在消费者代码中捕获异常，并将消息发送到 "dead-letter-topic"。示例：

try { consumer.poll(Duration.ofMillis(100)); } catch (Exception e) { producer.send(new ProducerRecord("dead-letter-topic", record.value())); }

在 RabbitMQ 中，将队列绑定到死信交换机。始终记录包含错误代码和时间戳等详细信息的错误日志。使用架构注册表验证事件并拒绝无效事件，例如通过 SchemaRegistryClient API。使用 Kafka 的 JMX 等工具监控延迟和错误；如果消费者延迟超过 1000 条消息，则设置告警。

具体使用示例

示例 1：基本的 Kafka 发布/订阅设置
要为用户事件设置发布/订阅：首先，创建一个主题：kafka-topics.sh --bootstrap-server localhost:9092 --create --topic user-events。然后生成一个事件：

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic user-events
{"userId": 1, "action": "login"}

接收该事件：kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic user-events --group mygroup。这样可以将生产者与消费者解耦，实现可扩展的事件处理。

示例 2：使用事件溯源实现 CQRS
对于电商应用，使用 Kafka 处理命令：通过 producer.send(new ProducerRecord("commands-topic", "{\"command\": \"placeOrder\", \"orderId\": 123}")) 向 "commands-topic" 发送消息。对于查询，通过消费事件并更新数据库来维护读取模型。消费者代码示例：

consumer.subscribe(Arrays.asList("events-topic"));
while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord<String, String> record : records) { updateReadModel(record.value()); } }

这能确保写操作与读操作分开处理，从而提升性能。

图关系

• 与集群相关：se-architecture
• 关联标签：event-driven、kafka、eventsourcing、cqrs
• 链接至：se-deployment（用于 Kafka 集群设置）、se-data-pipelines（用于事件流集成）