利用 NATS JetStream 与 InfluxDB 构建基于 BASE 原则的 GraphQL 事件溯源投影

分布式架构

文章字数: 3.8k

阅读时长: 17 分

技术痛点：失控的物联网设备状态管理

最初，我们为物联网（IoT）设备集群设计的状态管理服务是一个标准的 CRUD API，后端是 PostgreSQL。设备每秒上报一次心跳和多个传感器读数，API 直接执行 UPDATE 操作。这套方案在设备规模尚小的时候运行良好，但当设备数量突破一万台时，系统开始出现瓶颈。数据库的写并发竞争变得异常激烈，UPDATE 语句频繁锁表，导致心跳上报出现大量超时。更糟糕的是，我们丢失了所有历史状态。当需要排查某个设备在过去一小时内的温度变化趋势时，我们只能查询到最后一次的状态，历史数据踪迹全无。

这种设计在本质上是脆弱的。它试图用一个可变的、代表“当前状态”的记录来应对一个由连续、不可变事件组成的世界。我们需要一种新的架构，它不仅要解决写性能问题，还要能保留完整的状态变更历史，并且为未来的数据分析需求提供支撑。

初步构想：事件溯源与CQRS的引入

我们决定转向事件溯源（Event Sourcing, ES）架构。核心思路很简单：不再存储设备当前的状态，而是存储导致状态变更的一系列事件。例如，不再执行 UPDATE devices SET temperature = 25.5 WHERE id = 'xyz', 而是记录一个 DeviceTemperatureChanged 事件，内容为 {deviceId: 'xyz', temperature: 25.5, timestamp: ...}。所有事件被持久化到一个只允许追加的日志（Append-only Log）中。任何设备在任意时间点的状态，都可以通过重放其相关事件流来重建。

这种模式天然地分离了读写操作，非常适合引入命令查询职责分离（Command Query Responsibility Segregation, CQRS）。

写模型（Command Side）: 极其简单。接收到设备上报后，生成一个事件，然后把它扔进事件日志里。这个操作速度极快，几乎没有并发冲突。
读模型（Query Side）: 异步地消费事件日志，将事件“投影”（Project）成一个或多个为查询优化的数据模型。

这个架构的核心是BASE原则（Basically Available, Soft state, Eventually consistent）。我们接受读模型的数据会存在毫秒级的延迟，以此换取整个系统的高可用性（BA）和极高的写入吞吐量。

技术选型决策：NATS, InfluxDB 与 GraphQL

事件日志 - NATS JetStream: 我们需要一个高性能、持久化的消息系统。Kafka 是一个常见的选项，但在我们的场景中，运维复杂度偏高。NATS JetStream 提供了我们需要的一切：基于 Raft 的持久化、At-Least-Once 投递保证、简单的部署和运维，以及极高的性能。它作为事件日志的载体再合适不过。
读模型投影 - InfluxDB: 我们的核心查询场景是时序分析：某个设备在一段时间内的指标变化、一组设备的平均功耗等。关系型数据库无法高效处理这类查询。InfluxDB 作为一款专为时序数据设计的数据库，其数据模型（tags, fields, timestamp）与我们的事件天然契合。设备ID可以作为 tag，各类传感器读数作为 fields。
查询API - GraphQL: 前端和数据分析团队需要灵活地查询数据。他们可能想一次性获取某个设备的温度和湿度历史，或者查询多个设备在特定时间范围内的平均电压。如果使用 REST API，我们将需要开发大量的 endpoint。GraphQL 让客户端可以精确地声明自己需要的数据，一次请求获取所有信息，完美契合我们复杂的查询需求。

整体架构数据流如下：

flowchart TD
    subgraph "写模型 (Command Side)"
        A[IoT Device] -- gRPC/HTTP --> B(State Update Service);
        B -- Publish Event --> C{NATS JetStream Stream};
    end

    subgraph "读模型 (Query Side)"
        D(Projector Service) -- Durable Consumer --> C;
        D -- Write Points --> E[(InfluxDB)];
        F(GraphQL Service) -- Flux Query --> E;
        G[Client/Dashboard] -- GraphQL Query --> F;
    end

步骤化实现：构建高吞吐的投影管道

我们使用 Go 语言来实现整个后端服务。

1. 定义事件并发布到 NATS JetStream

首先，我们需要定义清晰的事件结构。在真实项目中，通常会使用 Protobuf 或 Avro 来保证 schema 的演进能力。这里为了简化，我们使用 JSON。

// internal/events/events.go
package events

import (
	"encoding/json"
	"time"
)

const (
	StreamName     = "DEVICE_EVENTS"
	StreamSubjects = "events.device.>"
)

// EventType 定义了事件类型
type EventType string

const (
	DeviceStatusChanged   EventType = "DeviceStatusChanged"
	DeviceReadingReported EventType = "DeviceReadingReported"
)

// EventHeader 包含所有事件的元数据
type EventHeader struct {
	EventID   string    `json:"eventId"`
	EventType EventType `json:"eventType"`
	DeviceID  string    `json:"deviceId"`
	Timestamp time.Time `json:"timestamp"`
}

// DeviceStatusChangedEvent 当设备在线/离线状态改变时触发
type DeviceStatusChangedEvent struct {
	Header EventHeader `json:"header"`
	Status string      `json:"status"` // e.g., "online", "offline"
}

// DeviceReadingReportedEvent 当设备上报传感器读数时触发
type DeviceReadingReportedEvent struct {
	Header   EventHeader          `json:"header"`
	Readings map[string]float64 `json:"readings"` // e.g., {"temperature": 25.5, "humidity": 60.1}
}

// ToJSON 序列化事件
func ToJSON(v interface{}) ([]byte, error) {
	return json.Marshal(v)
}

发布服务的核心逻辑是连接 NATS JetStream 并发布消息。这里的关键在于错误处理和发布选项的配置。

// cmd/publisher/main.go
package main

import (
	"context"
	"log"
	"os"
	"time"

	"github.com/google/uuid"
	"github.com/nats-io/nats.go"
	"github.com/yourapp/internal/events"
)

func main() {
	natsURL := os.Getenv("NATS_URL")
	if natsURL == "" {
		natsURL = nats.DefaultURL
	}

	nc, err := nats.Connect(natsURL)
	if err != nil {
		log.Fatalf("Error connecting to NATS: %v", err)
	}
	defer nc.Close()

	js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))
	if err != nil {
		log.Fatalf("Error getting JetStream context: %v", err)
	}

	// 确保 Stream 存在，生产环境中通常通过 IaC 工具管理
	_, err = js.AddStream(&nats.StreamConfig{
		Name:     events.StreamName,
		Subjects: []string{events.StreamSubjects},
	})
	if err != nil {
		log.Printf("Could not create stream, likely already exists: %v", err)
	}
    
    // 模拟一个设备上报数据
	deviceID := "device-007"
	publishDeviceReading(js, deviceID)
}

func publishDeviceReading(js nats.JetStreamContext, deviceID string) {
	event := events.DeviceReadingReportedEvent{
		Header: events.EventHeader{
			EventID:   uuid.New().String(),
			EventType: events.DeviceReadingReported,
			DeviceID:  deviceID,
			Timestamp: time.Now().UTC(),
		},
		Readings: map[string]float64{
			"temperature": 22.5 + (float64(time.Now().Second()) * 0.1),
			"humidity":    55.2 - (float64(time.Now().Second()) * 0.05),
			"voltage":     12.1,
		},
	}

	eventJSON, err := events.ToJSON(event)
	if err != nil {
		log.Printf("Error marshaling event: %v", err)
		return
	}

    // 主题格式: events.device.{deviceID}.{eventType}
	subject := "events.device." + deviceID + "." + string(events.DeviceReadingReported)

    // 使用 MsgId 来支持幂等性处理
	ack, err := js.Publish(subject, eventJSON, nats.MsgId(event.Header.EventID))
	if err != nil {
		log.Printf("Error publishing event: %v", err)
		return
	}

	log.Printf("Published event %s to stream %s, sequence %d", event.Header.EventID, ack.Stream, ack.Sequence)
}

这里的 nats.MsgId 是一个关键点。它为消息设置了一个唯一的ID，我们可以在消费端利用这个ID来实现幂等性，防止因为网络重试等原因导致同一事件被处理多次。

2. 实现投影服务：消费事件并写入 InfluxDB

投影服务是整个系统的核心。它必须是健壮的、可水平扩展的，并且能高效地将事件写入 InfluxDB。

// cmd/projector/main.go
package main

import (
	"context"
	"encoding/json"
	"log"
	"os"
	"time"

	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
	"github.com/influxdata/influxdb-client-go/v2/api"
	"github.com/nats-io/nats.go"
	"github.com/yourapp/internal/events"
)

const (
	durableConsumerName = "influxdb-projector"
	influxBucket        = "iot_device_data"
	influxOrg           = "my-org"
)

func main() {
	// --- InfluxDB Client Setup ---
	influxURL := os.Getenv("INFLUX_URL")
	influxToken := os.Getenv("INFLUX_TOKEN")
	influxClient := influxdb2.NewClient(influxURL, influxToken)
	defer influxClient.Close()
    
    // 使用带批处理的 WriteAPI 能够显著提升性能
	writeAPI := influxClient.WriteAPI(influxOrg, influxBucket)
	// 在程序退出时刷新所有挂起的写入
	defer writeAPI.Flush()
    // 监控写入错误
	go func() {
		for err := range writeAPI.Errors() {
			log.Printf("InfluxDB write error: %v", err)
		}
	}()

	// --- NATS JetStream Consumer Setup ---
	natsURL := os.Getenv("NATS_URL")
	nc, err := nats.Connect(natsURL)
	if err != nil {
		log.Fatalf("Error connecting to NATS: %v", err)
	}
	defer nc.Close()

	js, err := nc.JetStream()
	if err != nil {
		log.Fatalf("Error getting JetStream context: %v", err)
	}

	// 创建一个持久的、可分发的消费者（Durable Consumer）
    // 这是实现水平扩展的关键
	sub, err := js.PullSubscribe(events.StreamSubjects, durableConsumerName, nats.BindStream(events.StreamName))
	if err != nil {
		log.Fatalf("Error creating pull subscription: %v", err)
	}

	log.Println("Projector service started. Waiting for events...")

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

    // 循环拉取消息
	for {
		msgs, err := sub.Fetch(100, nats.MaxWait(10*time.Second)) // Batch fetch
		if err != nil {
            // 如果是超时，是正常行为，继续循环
			if err == nats.ErrTimeout {
				continue
			}
			log.Printf("Error fetching messages: %v", err)
			time.Sleep(1 * time.Second)
			continue
		}

		for _, msg := range msgs {
			processMessage(msg, writeAPI)
            // 手动确认消息，这是 At-Least-Once 语义的保证
            // 如果在 processMessage 中发生 panic 或服务崩溃，消息不会被确认，
            // 下次服务重启后会重新消费。
			msg.Ack()
		}
	}
}

func processMessage(msg *nats.Msg, writeAPI api.WriteAPI) {
	var header events.EventHeader
	if err := json.Unmarshal(msg.Data, &header); err != nil {
		log.Printf("Error unmarshaling event header for message on subject %s: %v", msg.Subject, err)
		return // 无法解析的消息，直接丢弃或移入死信队列
	}

	switch header.EventType {
	case events.DeviceReadingReported:
		var event events.DeviceReadingReportedEvent
		if err := json.Unmarshal(msg.Data, &event); err != nil {
			log.Printf("Error unmarshaling DeviceReadingReportedEvent: %v", err)
			return
		}
		handleDeviceReading(event, writeAPI)
	// ... 可以添加其他事件类型的处理
	default:
		log.Printf("Unknown event type: %s", header.EventType)
	}
}

func handleDeviceReading(event events.DeviceReadingReportedEvent, writeAPI api.WriteAPI) {
    // 这里的坑在于，如果一个事件包含多个读数，我们是写成一行还是多行？
    // 写成一行（多个 field）查询效率更高，但 schema 不够灵活。
    // 在真实项目中，我们可能会为每种类型的 reading 创建一个 measurement。
    // 这里为了演示，将所有 reading 作为 fields 写入一个 point。
	tags := map[string]string{
		"deviceId": event.Header.DeviceID,
	}
	
	p := influxdb2.NewPoint(
		"device_readings", // InfluxDB measurement
		tags,
		event.Readings, // map[string]float64 直接作为 fields
		event.Header.Timestamp,
	)

	writeAPI.WritePoint(p)
	log.Printf("Projected event %s for device %s", event.Header.EventID, event.Header.DeviceID)
}

一个常见的错误是在处理每条消息后都同步地写入数据库，这会带来巨大的 I/O 开销。influxdb-client-go 的 WriteAPI 内部实现了异步批处理，是生产环境下的最佳实践。我们只需要不断调用 WritePoint，客户端库会在后台收集数据点，批量发送。

使用 PullSubscribe 和 Fetch 也是一个重要的决策。与 Push 模式（服务器主动推送）相比，Pull 模式让消费者可以根据自己的处理能力来控制消息消费的速率，实现了天然的背压（backpressure），系统更加稳定。

3. 构建 GraphQL 服务查询 InfluxDB

最后一步是暴露一个 GraphQL 接口来查询我们投影好的数据。

首先，定义 GraphQL Schema。

# schema.graphql
schema {
    query: Query
}

type Query {
    deviceReadings(
        deviceId: String!,
        start: String!, 
        stop: String!,
        metrics: [String!]!
    ): [ReadingPoint]
}

type ReadingPoint {
    time: String!
    values: [MetricValue!]
}

type MetricValue {
    metric: String!
    value: Float!
}

接下来是 Go 服务实现。我们将使用 graphql-go 库，并编写解析器（Resolver）来执行 InfluxDB 的 Flux 查询。

// cmd/graphql/main.go
package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "os"
    "strings"
    "time"

    "github.com/graphql-go/graphql"
    "github.com/graphql-go/handler"
    influxdb2 "github.com/influxdata/influxdb-client-go/v2"
)

type GraphQLServer struct {
    influxClient influxdb2.Client
    schema       graphql.Schema
}

func NewGraphQLServer() (*GraphQLServer, error) {
    influxURL := os.Getenv("INFLUX_URL")
    influxToken := os.Getenv("INFLUX_TOKEN")
    client := influxdb2.NewClient(influxURL, influxToken)

    s := &GraphQLServer{influxClient: client}

    // ... GraphQL Schema 定义 (代码省略，使用上面的 schema.graphql 内容)
    // ... 这里需要将 schema.graphql 文本解析成 graphql-go 的对象

    fields := graphql.Fields{
        "deviceReadings": &graphql.Field{
            Type: graphql.NewList(readingPointType), // 定义在下面
            Args: graphql.FieldConfigArgument{
                "deviceId": &graphql.ArgumentConfig{Type: graphql.NewNonNull(graphql.String)},
                "start":    &graphql.ArgumentConfig{Type: graphql.NewNonNull(graphql.String)},
                "stop":     &graphql.ArgumentConfig{Type: graphql.NewNonNull(graphql.String)},
                "metrics":  &graphql.ArgumentConfig{Type: graphql.NewNonNull(graphql.NewList(graphql.String))},
            },
            Resolve: s.resolveDeviceReadings,
        },
    }

    rootQuery := graphql.ObjectConfig{Name: "RootQuery", Fields: fields}
    schemaConfig := graphql.SchemaConfig{Query: graphql.NewObject(rootQuery)}
    schema, err := graphql.NewSchema(schemaConfig)
    if err != nil {
        return nil, fmt.Errorf("failed to create new schema, error: %v", err)
    }
    s.schema = schema
    return s, nil
}


// resolveDeviceReadings 是核心的解析函数
func (s *GraphQLServer) resolveDeviceReadings(p graphql.ResolveParams) (interface{}, error) {
    // 从 GraphQL 参数中获取变量
	deviceID, _ := p.Args["deviceId"].(string)
	start, _ := p.Args["start"].(string)
	stop, _ := p.Args["stop"].(string)
	metricsArg, _ := p.Args["metrics"].([]interface{})

    var metrics []string
    for _, m := range metricsArg {
        metrics = append(metrics, m.(string))
    }

    // 动态构建 Flux 查询
    // 这里的坑在于如何安全地构建查询字符串，避免注入。
    // Flux 本身的设计对参数化支持较好，但客户端库的封装各异。
    // 在真实项目中，需要仔细审查以防止安全漏洞。
    
    // 生成 filter 部分: |> filter(fn: (r) => r._field == "metric1" or r._field == "metric2")
    fieldFilters := make([]string, len(metrics))
    for i, m := range metrics {
        fieldFilters[i] = fmt.Sprintf(`r._field == "%s"`, m)
    }
    filterStr := strings.Join(fieldFilters, " or ")

	fluxQuery := fmt.Sprintf(`
        from(bucket: "%s")
        |> range(start: %s, stop: %s)
        |> filter(fn: (r) => r._measurement == "device_readings")
        |> filter(fn: (r) => r.deviceId == "%s")
        |> filter(fn: (r) => %s)
        |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
    `, influxBucket, start, stop, deviceID, filterStr)

    log.Printf("Executing Flux query: %s", fluxQuery)

	queryAPI := s.influxClient.QueryAPI(influxOrg)
	result, err := queryAPI.Query(context.Background(), fluxQuery)
	if err != nil {
		return nil, err
	}

    // ... 解析 InfluxDB 返回的结果并塑造成 GraphQL Schema 定义的结构
    // (这部分代码较为繁琐，主要是数据结构转换，此处省略)
    // 最终返回一个 []ReadingPoint 类型的切片
	
    // 模拟返回数据
	type MetricValue struct {
		Metric string  `json:"metric"`
		Value  float64 `json:"value"`
	}
	type ReadingPoint struct {
		Time   string        `json:"time"`
		Values []MetricValue `json:"values"`
	}

	mockResult := []ReadingPoint{
		{
			Time: time.Now().Format(time.RFC3339),
			Values: []MetricValue{
				{Metric: "temperature", Value: 25.5},
				{Metric: "humidity", Value: 60.1},
			},
		},
	}

	return mockResult, nil
}


func main() {
    server, err := NewGraphQLServer()
    if err != nil {
        log.Fatalf("Failed to start server: %v", err)
    }

    h := handler.New(&handler.Config{
        Schema:   &server.schema,
        Pretty:   true,
        GraphiQL: true,
    })

    http.Handle("/graphql", h)
    log.Println("GraphQL server listening on :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

// 需要定义 GraphQL 类型
var metricValueType *graphql.Object
var readingPointType *graphql.Object

func init() {
    metricValueType = graphql.NewObject(graphql.ObjectConfig{
        Name: "MetricValue",
        Fields: graphql.Fields{
            "metric": &graphql.Field{Type: graphql.String},
            "value":  &graphql.Field{Type: graphql.Float},
        },
    })

    readingPointType = graphql.NewObject(graphql.ObjectConfig{
        Name: "ReadingPoint",
        Fields: graphql.Fields{
            "time":   &graphql.Field{Type: graphql.String},
            "values": &graphql.Field{Type: graphql.NewList(metricValueType)},
        },
    })
}

动态构建 Flux 查询是这里的核心。pivot 函数特别有用，它能将 InfluxDB 的行式存储格式转换成我们需要的列式结构（每个时间点一行，每个指标一列），这与 GraphQL 的返回结构更匹配。

当前方案的局限性与未来迭代

这套架构解决了最初的性能和数据历史问题，但它并非银弹。首先，我们完全拥抱了最终一致性。从事件发布到数据可在 GraphQL API 中查询到，存在一个可观测的延迟（通常是几十到几百毫ซ秒）。这个延迟对于我们的设备监控场景是完全可以接受的，但对于需要强一致性的场景（如交易系统）则不适用。

其次，投影服务的健壮性至关重要。如果投影服务宕机，读模型将停止更新。我们通过将 NATS Consumer 设置为 Durable 来保证服务重启后能从上次中断的地方继续消费，但仍需完备的监控和告警来确保其高可用。未来可以考虑将投影服务部署为 Kubernetes 中的多个副本，利用 NATS 的消费者分发机制实现负载均衡和高可用。

最后，事件 Schema 的演进是一个长期挑战。如果未来我们为 DeviceReadingReportedEvent 增加了一个新的字段，旧的投影服务代码可能无法解析新版事件。我们需要引入一套严格的 Schema 版本管理和演进策略，例如使用 Schema Registry，并让投影服务能够兼容处理多个版本的事件。

GraphQL NATS InfluxDB BASE Event Sourcing Go

基于Dart后端与Lit Web Components的事件驱动型微前端架构实现

2023-10-27 软件工程与架构

软件工程与架构 Dart Lit 事件驱动架构 esbuild

使用 gRPC-Go 构建 AI 模型流式推理服务并与 Laravel 集成

2023-10-27 后端架构

Laravel 微服务 gRPC-Go 模型部署异构系统