connector

package
v0.5.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Mar 29, 2026 License: MIT Imports: 20 Imported by: 0

README

connector

Go Reference

connector 是 Genesis 的 L1 基础设施层组件，负责管理与外部服务的原始连接。它为 Genesis 各上层组件提供统一的初始化、连接建立、健康检查和关闭语义，支持 Redis、MySQL、PostgreSQL、SQLite、Etcd、NATS、Kafka 七种后端。

组件定位

  • 两阶段初始化：New 验证配置，Connect 建立 I/O 连接，Fail-fast 且幂等可重试
  • 借用模型：连接器拥有底层连接生命周期，上层组件不应关闭连接器；它们的 Close 只处理自身资源，通常不接管底层连接
  • 类型安全：泛型接口 TypedConnector[T] 提供编译期类型检查，无运行时类型断言
  • 健康检查：HealthCheck 主动探测（有 I/O），IsHealthy 读取缓存（无 I/O）
  • 集成 clog，自动注入 namespace=connector 与连接器名称，方便按组件过滤日志

connector 不负责 ORM 查询、缓存策略、消息路由等业务逻辑，这些能力属于 L2 及以上组件。

快速开始

redisConn, err := connector.NewRedis(&connector.RedisConfig{
    Addr: "localhost:6379",
}, connector.WithLogger(logger))
if err != nil {
    return err
}
defer redisConn.Close()

if err := redisConn.Connect(ctx); err != nil {
    return err
}

client := redisConn.GetClient()
client.Set(ctx, "key", "value", time.Hour)

支持的连接器

类型 接口 底层客户端 工厂函数
Redis RedisConnector *redis.Client NewRedis
MySQL MySQLConnector *gorm.DB NewMySQL
PostgreSQL PostgreSQLConnector *gorm.DB NewPostgreSQL
SQLite SQLiteConnector *gorm.DB NewSQLite
Etcd EtcdConnector *clientv3.Client NewEtcd
NATS NATSConnector *nats.Conn NewNATS
Kafka KafkaConnector *kgo.Client NewKafka

推荐使用方式

资源所有权

连接器遵循"谁创建，谁释放"原则。上层组件（cache、dlock 等）不应调用连接器的 Close()，应用层通过 defer 按 LIFO 顺序释放：

redisConn, _ := connector.NewRedis(&cfg.Redis, connector.WithLogger(logger))
defer redisConn.Close()
redisConn.Connect(ctx)

// 注入连接器；cache 不拥有其生命周期
dist, _ := cache.NewDistributed(&cfg.Cache, cache.WithRedisConnector(redisConn))
健康检查

定期调用 HealthCheck 更新缓存状态，业务路径用 IsHealthy 快速判断：

go func() {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    for range ticker.C {
        if err := conn.HealthCheck(ctx); err != nil {
            logger.Warn("connector health check failed", clog.Error(err))
        }
    }
}()

错误处理

var (
    ErrConnection  = xerrors.New("connector: connection failed")
    ErrConfig      = xerrors.New("connector: invalid config")
    ErrHealthCheck = xerrors.New("connector: health check failed")
    ErrClientNil   = xerrors.New("connector: client is nil")
)

使用 xerrors.Is 匹配哨兵错误，ErrConnection 可重试，ErrConfig 是程序 bug 需修正配置。

测试

go test ./connector/... -count=1
go test -race ./connector/... -count=1

集成测试通过 testcontainers 自动启动容器，直接运行即可，无需手动配置 Docker 环境。

相关文档

Documentation

Overview

Package connector 为 Genesis 提供统一的连接管理能力。

核心特性：

  • 统一抽象：通过 Connector 接口提供一致的连接管理 API
  • 类型安全：通过 TypedConnector[T] 泛型接口确保编译时类型检查
  • 多数据源支持：Redis、MySQL、PostgreSQL、SQLite、Etcd、NATS、Kafka
  • 健康检查：提供主动探活和缓存态读取，但不负责统一重连或故障恢复
  • 并发安全：所有公开方法均为并发安全，支持多协程同时访问
  • 资源管理：遵循"谁创建，谁负责释放"原则，Close() 应在应用层调用

设计理念：

  • 接口优先：定义清晰的接口契约，实现细节可替换
  • 显式依赖注入：通过构造函数注入依赖，避免全局状态
  • 幂等连接：Connect() 方法可安全重复调用
  • 延迟连接：NewXXX() 创建连接器但不立即建立连接，Connect() 时才连接

基本使用： 

cfg := &connector.RedisConfig{
	Addr:     "127.0.0.1:6379",
	Password: "",
	DB:       0,
}
conn, err := connector.NewRedis(cfg, connector.WithLogger(logger))
if err != nil {
	panic(err)
}
defer conn.Close()

// 建立连接（幂等，可多次调用）
if err := conn.Connect(ctx); err != nil {
	panic(err)
}

// 获取类型安全的客户端
client := conn.GetClient()
result, err := client.Get(ctx, "key").Result()

资源所有权： 

Connector 拥有底层连接的生命周期，应通过 defer 确保 Close() 被调用。
Component（如 cache、dlock）仅借用 Connector，不应调用 Close()。
应用层应按照 LIFO 顺序释放资源：先关闭依赖 Connector 的组件，再关闭 Connector。

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrConnection 连接建立失败
	ErrConnection = xerrors.New("connector: connection failed")

	// ErrConfig 配置无效
	ErrConfig = xerrors.New("connector: invalid config")

	// ErrHealthCheck 健康检查失败
	ErrHealthCheck = xerrors.New("connector: health check failed")

	// ErrClientNil 客户端为空（未初始化或已关闭）
	ErrClientNil = xerrors.New("connector: client is nil")
)

Sentinel Errors - 连接器专用的哨兵错误

Functions

This section is empty.

Types

type Connector 

type Connector interface {
	// Connect 建立连接。
	//
	// 此方法是幂等的，可安全多次调用。首次调用时建立连接，
	// 后续调用直接返回 nil。连接过程阻塞直到成功或失败。
	//
	// 返回错误：
	//   - ErrConnection: 连接建立失败
	//   - ErrConfig: 配置无效
	Connect(ctx context.Context) error

	// Close 关闭连接并释放资源。
	//
	// 此方法是幂等的，可安全多次调用。关闭后，
	// GetClient() 将返回 nil，HealthCheck() 将返回 ErrClientNil。
	//
	// 重要：应在应用层通过 defer 确保调用，遵循"谁创建，谁负责释放"原则。
	Close() error

	// HealthCheck 检查连接健康状态。
	//
	// 通过发送测试请求验证连接可用性。此方法会更新内部健康状态缓存，
	// 可通过 IsHealthy() 快速读取。
	//
	// 返回错误：
	//   - ErrClientNil: 客户端未初始化或已关闭
	//   - ErrHealthCheck: 健康检查失败
	HealthCheck(ctx context.Context) error

	// IsHealthy 返回缓存的健康状态。
	//
	// 此方法无阻塞，直接返回最后一次 HealthCheck() 的结果。
	// 对于实时健康检查，应使用 HealthCheck() 方法。
	IsHealthy() bool

	// Name 返回连接实例名称。
	//
	// 名称用于日志记录和指标标识，应在配置中唯一标识此连接器实例。
	Name() string
}

Connector 定义所有连接器的通用行为。

所有连接器必须实现此接口，确保一致的连接管理体验。 接口方法均为并发安全，可从多个协程同时调用。

type EtcdConfig 

type EtcdConfig struct {
	// 基础配置
	Name string `mapstructure:"name" json:"name" yaml:"name"` // 连接器名称 (默认: "default")

	// 核心配置
	Endpoints []string `mapstructure:"endpoints" json:"endpoints" yaml:"endpoints"` // 连接地址列表 (必填)
	Username  string   `mapstructure:"username" json:"username" yaml:"username"`    // 认证用户 (可选)
	Password  string   `mapstructure:"password" json:"password" yaml:"password"`    // 认证密码 (可选)

	// 高级配置
	DialTimeout      time.Duration `mapstructure:"dial_timeout" json:"dial_timeout" yaml:"dial_timeout"`                   // 连接超时 (默认: 5s)
	KeepAliveTime    time.Duration `mapstructure:"keep_alive_time" json:"keep_alive_time" yaml:"keep_alive_time"`          // 心跳间隔 (默认: 10s)
	KeepAliveTimeout time.Duration `mapstructure:"keep_alive_timeout" json:"keep_alive_timeout" yaml:"keep_alive_timeout"` // 心跳超时 (默认: 3s)
}

EtcdConfig Etcd连接配置

type EtcdConnector 

type EtcdConnector interface {
	TypedConnector[*clientv3.Client]
}

EtcdConnector Etcd 连接器接口。

提供对 Etcd 键值存储的连接管理，支持分布式锁、配置中心、服务发现等场景。

func NewEtcd 

func NewEtcd(cfg *EtcdConfig, opts ...Option) (EtcdConnector, error)

NewEtcd 创建 Etcd 连接器 注意：实际连接在调用 Connect() 时建立

type KafkaConfig 

type KafkaConfig struct {
	// 基础配置
	Name string `mapstructure:"name" json:"name" yaml:"name"` // 连接器名称 (默认: "default")

	// 核心配置
	Seed []string `mapstructure:"seed" json:"seed" yaml:"seed"` // 初始连接节点 (必填)

	// 认证配置
	User     string `mapstructure:"user" json:"user" yaml:"user"`                // SASL 用户名 (可选)
	Password string `mapstructure:"password" json:"password" yaml:"password"`    // SASL 密码 (可选)
	ClientID string `mapstructure:"client_id" json:"client_id" yaml:"client_id"` // 客户端 ID (默认: "genesis-connector")

	// 连接配置
	ConnectTimeout       time.Duration `mapstructure:"connect_timeout" json:"connect_timeout" yaml:"connect_timeout"`                         // 连接超时 (默认: 10s)
	RequestTimeout       time.Duration `mapstructure:"request_timeout" json:"request_timeout" yaml:"request_timeout"`                         // 请求超时 (默认: 10s)
	AllowAutoTopicCreate bool          `mapstructure:"allow_auto_topic_create" json:"allow_auto_topic_create" yaml:"allow_auto_topic_create"` // 允许自动创建 Topic (默认: false)
}

KafkaConfig Kafka连接配置

type KafkaConnector 

type KafkaConnector interface {
	TypedConnector[*kgo.Client]
}

KafkaConnector Kafka 连接器接口。

提供对 Kafka 消息队列的连接管理，支持高吞吐的消息生产和消费。 基于 franz-go 客户端，提供现代的 Kafka 消费者组 API。

func NewKafka 

func NewKafka(cfg *KafkaConfig, opts ...Option) (KafkaConnector, error)

NewKafka 创建 Kafka 连接器 注意：实际连接在调用 Connect() 时建立

type MySQLConfig 

type MySQLConfig struct {
	// 基础配置
	Name string `mapstructure:"name" json:"name" yaml:"name"` // 连接器名称 (默认: "default")

	// 核心配置
	DSN      string `mapstructure:"dsn" json:"dsn" yaml:"dsn"`                // 完整 DSN (可选，若提供则忽略 Host/Port 等，优先级最高)
	Host     string `mapstructure:"host" json:"host" yaml:"host"`             // 主机地址 (DSN 未设置时必填)
	Port     int    `mapstructure:"port" json:"port" yaml:"port"`             // 端口 (默认: 3306)
	Username string `mapstructure:"username" json:"username" yaml:"username"` // 用户名 (DSN 未设置时必填)
	Password string `mapstructure:"password" json:"password" yaml:"password"` // 密码
	Database string `mapstructure:"database" json:"database" yaml:"database"` // 数据库名 (DSN 未设置时必填)

	// 高级配置
	Charset         string        `mapstructure:"charset" json:"charset" yaml:"charset"`                               // 字符集 (默认: "utf8mb4")
	MaxIdleConns    int           `mapstructure:"max_idle_conns" json:"max_idle_conns" yaml:"max_idle_conns"`          // 最大空闲连接数 (默认: 10)
	MaxOpenConns    int           `mapstructure:"max_open_conns" json:"max_open_conns" yaml:"max_open_conns"`          // 最大打开连接数 (默认: 100)
	ConnMaxLifetime time.Duration `mapstructure:"conn_max_lifetime" json:"conn_max_lifetime" yaml:"conn_max_lifetime"` // 连接最大生命周期 (默认: 1h)
	ConnectTimeout  time.Duration `mapstructure:"connect_timeout" json:"connect_timeout" yaml:"connect_timeout"`       // 连接超时 (默认: 5s)
}

MySQLConfig MySQL连接配置

type MySQLConnector 

type MySQLConnector interface {
	TypedConnector[*gorm.DB]
}

MySQLConnector MySQL 连接器接口。

提供对 MySQL 数据库的连接管理，基于 GORM ORM 框架。 支持连接池、预处理缓存、自动重连等特性。

func NewMySQL 

func NewMySQL(cfg *MySQLConfig, opts ...Option) (MySQLConnector, error)

NewMySQL 创建 MySQL 连接器 注意：实际连接在调用 Connect() 时建立

type NATSConfig 

type NATSConfig struct {
	// 基础配置
	Name string `mapstructure:"name" json:"name" yaml:"name"` // 连接器名称 (默认: "default")

	// 核心配置
	URL      string `mapstructure:"url" json:"url" yaml:"url"`                // 连接地址 (必填)，如 "nats://127.0.0.1:4222"
	Username string `mapstructure:"username" json:"username" yaml:"username"` // 用户名 (可选)
	Password string `mapstructure:"password" json:"password" yaml:"password"` // 密码 (可选)
	Token    string `mapstructure:"token" json:"token" yaml:"token"`          // 令牌 (可选)

	// 高级配置
	ConnectTimeout time.Duration `mapstructure:"connect_timeout" json:"connect_timeout" yaml:"connect_timeout"` // 连接超时 (默认: 5s)
	MaxReconnects  int           `mapstructure:"max_reconnects" json:"max_reconnects" yaml:"max_reconnects"`    // 最大重连次数 (默认: 60)
	ReconnectWait  time.Duration `mapstructure:"reconnect_wait" json:"reconnect_wait" yaml:"reconnect_wait"`    // 重连等待时间 (默认: 2s)
	PingInterval   time.Duration `mapstructure:"ping_interval" json:"ping_interval" yaml:"ping_interval"`       // ping间隔 (默认: 2m)
}

NATSConfig NATS连接配置

type NATSConnector 

type NATSConnector interface {
	TypedConnector[*nats.Conn]
}

NATSConnector NATS 连接器接口。

提供对 NATS 消息系统的连接管理，支持发布订阅、请求响应、消息队列等模式。 内置自动重连机制，网络故障时会自动尝试恢复连接。

func NewNATS 

func NewNATS(cfg *NATSConfig, opts ...Option) (NATSConnector, error)

NewNATS 创建 NATS 连接器 注意：实际连接在调用 Connect() 时建立

type Option 

type Option func(*options)

Option 配置连接器的选项

func WithLogger 

func WithLogger(logger clog.Logger) Option

type PostgreSQLConfig 

type PostgreSQLConfig struct {
	// 基础配置
	Name string `mapstructure:"name" json:"name" yaml:"name"` // 连接器名称 (默认: "default")

	// 核心配置
	DSN      string `mapstructure:"dsn" json:"dsn" yaml:"dsn"`                // 完整 DSN (可选，若提供则忽略 Host/Port 等，优先级最高)
	Host     string `mapstructure:"host" json:"host" yaml:"host"`             // 主机地址 (DSN 未设置时必填)
	Port     int    `mapstructure:"port" json:"port" yaml:"port"`             // 端口 (默认: 5432)
	Username string `mapstructure:"username" json:"username" yaml:"username"` // 用户名 (DSN 未设置时必填)
	Password string `mapstructure:"password" json:"password" yaml:"password"` // 密码
	Database string `mapstructure:"database" json:"database" yaml:"database"` // 数据库名 (DSN 未设置时必填)

	// 高级配置
	SSLMode         string        `mapstructure:"sslmode" json:"sslmode" yaml:"sslmode"`                               // SSL 模式 (默认: "disable")
	MaxIdleConns    int           `mapstructure:"max_idle_conns" json:"max_idle_conns" yaml:"max_idle_conns"`          // 最大空闲连接数 (默认: 10)
	MaxOpenConns    int           `mapstructure:"max_open_conns" json:"max_open_conns" yaml:"max_open_conns"`          // 最大打开连接数 (默认: 100)
	ConnMaxLifetime time.Duration `mapstructure:"conn_max_lifetime" json:"conn_max_lifetime" yaml:"conn_max_lifetime"` // 连接最大生命周期 (默认: 1h)
	ConnectTimeout  time.Duration `mapstructure:"connect_timeout" json:"connect_timeout" yaml:"connect_timeout"`       // 连接超时 (默认: 5s)
	Timezone        string        `mapstructure:"timezone" json:"timezone" yaml:"timezone"`                            // 时区 (默认: "UTC")
}

PostgreSQLConfig PostgreSQL连接配置

type PostgreSQLConnector 

type PostgreSQLConnector interface {
	TypedConnector[*gorm.DB]
}

PostgreSQLConnector PostgreSQL 连接器接口。

提供对 PostgreSQL 数据库的连接管理，基于 GORM ORM 框架。 支持高级数据类型（JSONB、ARRAY、GIS）、复杂查询、全文搜索等企业级特性。

func NewPostgreSQL 

func NewPostgreSQL(cfg *PostgreSQLConfig, opts ...Option) (PostgreSQLConnector, error)

NewPostgreSQL 创建 PostgreSQL 连接器 注意：实际连接在调用 Connect() 时建立

type RedisConfig 

type RedisConfig struct {
	// 基础配置
	Name string `mapstructure:"name" json:"name" yaml:"name"` // 连接器名称 (默认: "default")

	// 核心配置
	Addr     string `mapstructure:"addr" json:"addr" yaml:"addr"`             // 连接地址 (必填)，如 "127.0.0.1:6379"
	Password string `mapstructure:"password" json:"password" yaml:"password"` // 认证密码 (可选)
	DB       int    `mapstructure:"db" json:"db" yaml:"db"`                   // 数据库编号 (默认: 0)

	// 高级配置
	PoolSize     int           `mapstructure:"pool_size" json:"pool_size" yaml:"pool_size"`                // 连接池大小 (默认: 10)
	MinIdleConns int           `mapstructure:"min_idle_conns" json:"min_idle_conns" yaml:"min_idle_conns"` // 最小空闲连接数 (默认: 5)
	DialTimeout  time.Duration `mapstructure:"dial_timeout" json:"dial_timeout" yaml:"dial_timeout"`       // 连接超时 (默认: 5s)
	ReadTimeout  time.Duration `mapstructure:"read_timeout" json:"read_timeout" yaml:"read_timeout"`       // 读取超时 (默认: 3s)
	WriteTimeout time.Duration `mapstructure:"write_timeout" json:"write_timeout" yaml:"write_timeout"`    // 写入超时 (默认: 3s)

	// 可观测性
	EnableTracing bool `mapstructure:"enable_tracing" json:"enable_tracing" yaml:"enable_tracing"` // 是否启用 Tracing (透传给 redisotel)
}

RedisConfig Redis连接配置

type RedisConnector 

type RedisConnector interface {
	TypedConnector[*redis.Client]
}

RedisConnector Redis 连接器接口。

提供对 Redis 服务器的连接管理，支持连接池、Pipeline、事务等特性。

func NewRedis 

func NewRedis(cfg *RedisConfig, opts ...Option) (RedisConnector, error)

NewRedis 创建 Redis 连接器 注意：实际连接在调用 Connect() 时建立

type SQLiteConfig 

type SQLiteConfig struct {
	// 基础配置
	Name string `mapstructure:"name" json:"name" yaml:"name"` // 连接器名称 (默认: "default")

	// 核心配置
	Path string `mapstructure:"path" json:"path" yaml:"path"` // 数据库文件路径 (必填)，如 "./test.db" 或 "file::memory:?cache=shared"
}

SQLiteConfig SQLite连接配置

type SQLiteConnector 

type SQLiteConnector interface {
	TypedConnector[*gorm.DB]
}

SQLiteConnector SQLite 连接器接口。

提供对 SQLite 数据库的连接管理，基于 GORM ORM 框架。 支持内存数据库和文件数据库，适合测试和嵌入式场景。

func NewSQLite 

func NewSQLite(cfg *SQLiteConfig, opts ...Option) (SQLiteConnector, error)

NewSQLite 创建 SQLite 连接器 注意：实际连接在调用 Connect() 时建立

type TypedConnector 

type TypedConnector[T any] interface {
	Connector

	// GetClient 返回底层客户端实例。
	//
	// 调用者应通过此客户端执行实际的数据操作。
	// 注意：在 Connect() 之前或 Close() 之后调用可能返回 nil。
	GetClient() T
}

TypedConnector 提供类型安全的客户端访问。

此接口组合了 Connector 基础接口，并添加了 GetClient() 方法 用于获取特定类型的客户端。所有具体连接器接口都应基于此定义。

类型参数 T 是客户端类型，如 *redis.Client、*gorm.DB 等。

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.