跳到主要内容

从零构建用户 RPC 服务

这次实战要完成什么？​

  • 结果: 创建用户、按 ID 查询，通过真实 RPC 获取业务异常;
  • 顺序: 从空目录按节保存文件，完成全部实现后执行测试；带路径的代码块就是该文件的完整内容，父目录不存在时先创建;
  • 基线: Go 1.26，Kitex/tool v0.16.3、thriftgo v0.4.5，GORM v1.31.1、MySQL driver v1.6.0、MySQL 8.4;
  • 篇幅: 连续可执行实战按本轮要求豁免 144 行和平均 55 行限制，保留必要的完整代码;
  • 边界: 示例固定端口 8888，两个项目分别运行；邮箱仅做基础检查，创建不启用业务自动重试;

如何准备项目和工具？​

New-Item -ItemType Directory user-service
Set-Location user-service
New-Item -ItemType Directory idl
go install github.com/cloudwego/kitex/tool/cmd/kitex@v0.16.3
go install github.com/cloudwego/thriftgo@v0.4.5
kitex --version
go version
  • PATH: 工具命令不可用时，把 Go bin 目录加入当前终端 PATH，再执行版本检查;
  • 依赖: 先保存下面的 go.mod，完成文件后 go mod tidy 会补齐间接依赖与 go.sum;
go.mod
module example.com/user-service

go 1.26.0

require (
 github.com/cloudwego/kitex v0.16.3
 gorm.io/gorm v1.31.1
 gorm.io/driver/mysql v1.6.0
)

如何定义接口并生成代码？​

idl/user.thrift
namespace go user

struct User {
  1: required i64 id
  2: required string email
  3: required string name
  4: required i64 created_at_unix
}

struct CreateUserRequest {
  1: required string email
  2: required string name
}

struct GetUserRequest {
  1: required i64 id
}

service UserService {
  User CreateUser(1: CreateUserRequest req)
  User GetUser(1: GetUserRequest req)
}
kitex -module example.com/user-service idl/user.thrift
  • 检查: 生成 kitex_gen/user 和 userservice；当前版本将 id 生成为 Id，后面的代码按实际生成字段使用;
  • 所有权: kitex_gen 不手改，业务放 internal 和 cmd;
  • 更新: 修改 IDL 后重复相同命令，检查生成差异后测试;

如何定义业务数据与错误？​

internal/user/entity.go
package user

import (
	"errors"
	"time"
)

var (
	ErrInvalidInput  = errors.New("invalid user input")
	ErrEmailConflict = errors.New("email conflict")
	ErrNotFound      = errors.New("user not found")
)

type User struct {
	ID        int64
	Email     string
	Name      string
	CreatedAt time.Time
}

type CreateInput struct {
	Email string
	Name  string
}
  • 数据: 领域对象不包含 HTTP、RPC 或 GORM 类型；输入错误、不存在和冲突由 Handler 映射为公开业务码;

如何声明存储接口？​

internal/user/repository.go
package user

import (
	"context"
)

type Repository interface {
	Create(ctx context.Context, user *User) error
	FindByID(ctx context.Context, id int64) (*User, error)
}
  • 接口: 只暴露业务需要的创建和查询操作，Service 不依赖 GORM;

如何实现业务服务？​

internal/user/service.go
package user

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"
)

type Service struct{ repo Repository }

func NewService(repo Repository) (*Service, error) {
	if repo == nil {
		return nil, errors.New("repository is required")
	}
	return &Service{repo: repo}, nil
}

func (s *Service) Create(ctx context.Context, in CreateInput) (*User, error) {
	in.Email = strings.ToLower(strings.TrimSpace(in.Email))
	in.Name = strings.TrimSpace(in.Name)
	if in.Email == "" || !strings.Contains(in.Email, "@") || len(in.Email) > 254 || in.Name == "" || len([]rune(in.Name)) > 100 {
		return nil, ErrInvalidInput
	}
	entity := &User{Email: in.Email, Name: in.Name, CreatedAt: time.Now().UTC()}
	if err := s.repo.Create(ctx, entity); err != nil {
		return nil, fmt.Errorf("create user: %w", err)
	}
	return entity, nil
}

func (s *Service) Get(ctx context.Context, id int64) (*User, error) {
	if id <= 0 {
		return nil, ErrInvalidInput
	}
	entity, err := s.repo.FindByID(ctx, id)
	if err != nil {
		return nil, fmt.Errorf("get user %d: %w", id, err)
	}
	return entity, nil
}
  • 输入: 检查字段长度与正数 ID；文章标题按 Unicode 字符计数，正文按字节限制；作者不可变，更新和删除先检查所有权;

如何用 GORM 实现存储？​

internal/user/repository_gorm.go
package user

import (
	"context"
	"errors"
	"fmt"
	"gorm.io/gorm"
	"time"
)

type userRow struct {
	ID        int64     `gorm:"primaryKey"`
	Email     string    `gorm:"size:254;not null;uniqueIndex"`
	Name      string    `gorm:"size:100;not null"`
	CreatedAt time.Time `gorm:"not null"`
}

type GormRepository struct{ db *gorm.DB }

func NewGormRepository(db *gorm.DB) (*GormRepository, error) {
	if db == nil {
		return nil, errors.New("database is required")
	}
	return &GormRepository{db: db}, nil
}

func (r *GormRepository) Create(ctx context.Context, entity *User) error {
	row := userRow{Email: entity.Email, Name: entity.Name, CreatedAt: entity.CreatedAt}
	if err := r.db.WithContext(ctx).Create(&row).Error; err != nil {
		if errors.Is(err, gorm.ErrDuplicatedKey) {
			return ErrEmailConflict
		}
		return fmt.Errorf("insert user: %w", err)
	}
	entity.ID = row.ID
	return nil
}

func (r *GormRepository) FindByID(ctx context.Context, id int64) (*User, error) {
	var row userRow
	err := r.db.WithContext(ctx).First(&row, id).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, ErrNotFound
	}
	if err != nil {
		return nil, fmt.Errorf("select user %d: %w", id, err)
	}
	return toEntity(row), nil
}
func (userRow) TableName() string { return "users" }
func toEntity(row userRow) *User {
	return &User{ID: row.ID, Email: row.Email, Name: row.Name, CreatedAt: row.CreatedAt}
}
  • 表名: 使用 TableName 对齐 SQL 的 users/articles，避免被推导为 user_rows/article_rows;
  • 错误: First 无记录转为领域错误；用户唯一键通过 TranslateError 转为 ErrDuplicatedKey;

如何建立数据表？​

migrations/001_create_users.sql
CREATE TABLE users (
  id BIGINT NOT NULL AUTO_INCREMENT,
  email VARCHAR(254) NOT NULL,
  name VARCHAR(100) NOT NULL,
  created_at DATETIME(3) NOT NULL,
  PRIMARY KEY (id),
  UNIQUE KEY uk_users_email (email)
);
  • 数据约束: email 唯一键处理并发重复创建，时间统一 UTC;
  • 迁移: Server 不自动建表，先执行后面的独立命令;

如何读取配置并连接 MySQL？​

internal/platform/database/open.go
package database

import (
	"context"
	"database/sql"
	"errors"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"os"
	"time"
)

func Open() (*gorm.DB, *sql.DB, error) {
	dsn := os.Getenv("USER_DATABASE_DSN")
	if dsn == "" {
		return nil, nil, errors.New("USER_DATABASE_DSN is required")
	}
	db, err := gorm.Open(mysql.Open(dsn), &gorm.Config{TranslateError: true})
	if err != nil {
		return nil, nil, err
	}
	pool, err := db.DB()
	if err != nil {
		return nil, nil, err
	}
	pool.SetMaxOpenConns(10)
	pool.SetMaxIdleConns(5)
	pool.SetConnMaxLifetime(3 * time.Minute)
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	if err = pool.PingContext(ctx); err != nil {
		pool.Close()
		return nil, nil, err
	}
	return db, pool, nil
}
  • 配置: USER_DATABASE_DSN 必填，密码和完整 DSN 不进入普通日志;
  • 池: max open=10、max idle=5 是本地示例值；Ping 有 3 秒预算，退出时关闭 pool;

如何在启动前执行迁移？​

cmd/migrate/main.go
package main

import (
	"context"
	"database/sql"
	"errors"
	"example.com/user-service/internal/platform/database"
	"log"
	"os"
	"time"
)

func run() error {
	_, pool, err := database.Open()
	if err != nil {
		return err
	}
	defer pool.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	conn, err := pool.Conn(ctx)
	if err != nil {
		return err
	}
	defer conn.Close()
	var locked int
	if err = conn.QueryRowContext(ctx, "SELECT GET_LOCK('notes_schema_migration', 10)").Scan(&locked); err != nil {
		return err
	}
	if locked != 1 {
		return errors.New("migration lock unavailable")
	}
	defer func() {
		release, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		conn.ExecContext(release, "SELECT RELEASE_LOCK('notes_schema_migration')")
	}()
	if _, err = conn.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS schema_migrations (version BIGINT PRIMARY KEY)"); err != nil {
		return err
	}
	var version int
	err = conn.QueryRowContext(ctx, "SELECT version FROM schema_migrations WHERE version=1").Scan(&version)
	if err == nil {
		return nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	ddl, err := os.ReadFile("migrations/001_create_users.sql")
	if err != nil {
		return err
	}
	if _, err = conn.ExecContext(ctx, string(ddl)); err != nil {
		return err
	}
	_, err = conn.ExecContext(ctx, "INSERT INTO schema_migrations(version) VALUES (1)")
	return err
}
func main() {
	if err := run(); err != nil {
		log.Fatal(err)
	}
}
  • 执行位置: 从项目根目录运行，读取 migrations 下的 SQL；版本 1 成功后重跑会跳过;
  • 并发: 命名锁与 DDL 使用同一连接，避免两个初始化进程同时建表;
  • 中断恢复: MySQL DDL 不随普通事务回滚；若建表成功但版本记录失败，需要人工核对表结构再修复版本记录，不能宣称自动恢复;
  • 后续开发: 使用团队统一的迁移工具，本命令只演示初始化;

如何把 RPC 请求交给业务服务？​

internal/transport/kitex/handler.go
package kitextransport

import (
	"context"
	"errors"
	userdomain "example.com/user-service/internal/user"
	"example.com/user-service/kitex_gen/user"
	"fmt"
	"github.com/cloudwego/kitex/pkg/kerrors"
)

type UserServiceImpl struct {
	service *userdomain.Service
}

func NewUserServiceImpl(service *userdomain.Service) (*UserServiceImpl, error) {
	if service == nil {
		return nil, errors.New("user service is required")
	}
	return &UserServiceImpl{service: service}, nil
}

func (h *UserServiceImpl) CreateUser(ctx context.Context, req *user.CreateUserRequest) (*user.User, error) {
	if req == nil {
		return nil, kerrors.NewBizStatusError(40001, "request is required")
	}
	entity, err := h.service.Create(ctx, userdomain.CreateInput{Email: req.Email, Name: req.Name})
	switch {
	case errors.Is(err, userdomain.ErrInvalidInput):
		return nil, kerrors.NewBizStatusError(40001, "invalid user")
	case errors.Is(err, userdomain.ErrEmailConflict):
		return nil, kerrors.NewBizStatusError(40901, "email already exists")
	case err != nil:
		return nil, fmt.Errorf("create user: %w", err)
	default:
		return toRPCUser(entity), nil
	}
}

func (h *UserServiceImpl) GetUser(ctx context.Context, req *user.GetUserRequest) (*user.User, error) {
	if req == nil {
		return nil, kerrors.NewBizStatusError(40001, "request is required")
	}
	entity, err := h.service.Get(ctx, req.Id)
	switch {
	case errors.Is(err, userdomain.ErrInvalidInput):
		return nil, kerrors.NewBizStatusError(40001, "invalid user id")
	case errors.Is(err, userdomain.ErrNotFound):
		return nil, kerrors.NewBizStatusError(40401, "user not found")
	case err != nil:
		return nil, fmt.Errorf("get user: %w", err)
	default:
		return toRPCUser(entity), nil
	}
}
func toRPCUser(u *userdomain.User) *user.User {
	return &user.User{Id: u.ID, Email: u.Email, Name: u.Name, CreatedAtUnix: u.CreatedAt.Unix()}
}
  • 适配: 输入检查、Service 调用、输出转换分别完成，业务错误码保持稳定；完整错误仅在服务端记录;

如何组装并启动进程？​

cmd/server/main.go
package main

import (
	"example.com/user-service/internal/platform/database"
	kitextransport "example.com/user-service/internal/transport/kitex"
	userdomain "example.com/user-service/internal/user"
	"example.com/user-service/kitex_gen/user/userservice"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/transmeta"
	"github.com/cloudwego/kitex/server"
	"log"
	"net"
	"os"
	"time"
)

func run() error {
	db, pool, err := database.Open()
	if err != nil {
		return err
	}
	defer pool.Close()
	repo, err := userdomain.NewGormRepository(db)
	if err != nil {
		return err
	}
	svc, err := userdomain.NewService(repo)
	if err != nil {
		return err
	}
	handler, err := kitextransport.NewUserServiceImpl(svc)
	if err != nil {
		return err
	}
	address := os.Getenv("USER_SERVICE_ADDRESS")
	if address == "" {
		address = ":8888"
	}
	addr, err := net.ResolveTCPAddr("tcp", address)
	if err != nil {
		return err
	}
	svr := userservice.NewServer(handler, server.WithServiceAddr(addr),
		server.WithServerBasicInfo(&rpcinfo.EndpointBasicInfo{ServiceName: "user-service"}),
		server.WithExitWaitTime(5*time.Second), server.WithEnableContextTimeout(true),
		server.WithMetaHandler(transmeta.ServerTTHeaderHandler))
	return svr.Run()
}
func main() {
	if err := run(); err != nil {
		log.Fatal(err)
	}
}
  • 入口: cmd/server 初始化 DB、Repository、Service、Handler，再创建 Server；TTHeader 和服务端 context 超时显式启用;
  • 退出: Run 返回后关闭 DB；实际部署按需补注册发现与 readiness;
  • 边界: 本例使用本地直连，不要求先部署治理平台;

如何用 Client 创建并查询用户？​

cmd/client/main.go
package main

import (
	"context"
	"example.com/user-service/kitex_gen/user"
	"example.com/user-service/kitex_gen/user/userservice"
	"fmt"
	"github.com/cloudwego/kitex/client"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"github.com/cloudwego/kitex/pkg/transmeta"
	"github.com/cloudwego/kitex/transport"
	"log"
	"time"
)

func main() {
	cli, err := userservice.NewClient(
		"user-service",
		client.WithHostPorts("127.0.0.1:8888"),
		client.WithConnectTimeout(300*time.Millisecond),
		client.WithRPCTimeout(time.Second),
		client.WithTransportProtocol(transport.TTHeader),
		client.WithMetaHandler(transmeta.ClientTTHeaderHandler),
	)
	if err != nil {
		log.Fatalf("create user client: %v", err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	created, err := cli.CreateUser(ctx, &user.CreateUserRequest{Email: fmt.Sprintf("alice-%d@example.com", time.Now().UnixNano()), Name: "Alice"})
	cancel()
	if err != nil {
		if bizErr, ok := kerrors.FromBizStatusError(err); ok {
			log.Fatalf("create user rejected: code=%d message=%s", bizErr.BizStatusCode(), bizErr.BizMessage())
		}
		log.Fatalf("create user: %v", err)
	}
	found, err := cli.GetUser(context.Background(), &user.GetUserRequest{Id: created.Id})
	if err != nil {
		log.Fatalf("get user: %v", err)
	}
	log.Printf("user=%d %s %s", found.Id, found.Email, found.Name)
}
  • 传输: 双方 TTHeader 与 MetaHandler 配套，Client 才能识别 BizStatus;
  • 运行: 每次生成不同邮箱，便于重复验证成功路径；业务 Client 在进程内复用;

如何不依赖 MySQL 验证业务？​

internal/user/service_test.go
package user

import (
	"context"
	"errors"
	"testing"
)

type fakeRepository struct{ called bool }

func (r *fakeRepository) Create(context.Context, *User) error {
	r.called = true
	return ErrEmailConflict
}
func (r *fakeRepository) FindByID(context.Context, int64) (*User, error) { return nil, ErrNotFound }
func TestServiceRejectsInvalidInput(t *testing.T) {
	repo := &fakeRepository{}
	svc, _ := NewService(repo)
	_, err := svc.Create(context.Background(), CreateInput{Email: "invalid", Name: "Alice"})
	if !errors.Is(err, ErrInvalidInput) || repo.called {
		t.Fatalf("invalid input reached repository: %v", err)
	}
	_, err = svc.Create(context.Background(), CreateInput{Email: "alice@example.com", Name: "Alice"})
	if !errors.Is(err, ErrEmailConflict) {
		t.Fatalf("lost conflict: %v", err)
	}
	_, err = svc.Get(context.Background(), 1)
	if !errors.Is(err, ErrNotFound) {
		t.Fatalf("lost not found: %v", err)
	}
}
internal/transport/kitex/handler_test.go
package kitextransport

import (
	"context"
	domain "example.com/user-service/internal/user"
	"example.com/user-service/kitex_gen/user"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"testing"
)

type missingRepo struct{}

func (missingRepo) Create(context.Context, *domain.User) error { return domain.ErrEmailConflict }
func (missingRepo) FindByID(context.Context, int64) (*domain.User, error) {
	return nil, domain.ErrNotFound
}
func TestBusinessCodes(t *testing.T) {
	svc, _ := domain.NewService(missingRepo{})
	h, _ := NewUserServiceImpl(svc)
	for _, tc := range []struct {
		req  *user.GetUserRequest
		code int32
	}{{nil, 40001}, {&user.GetUserRequest{Id: 1}, 40401}} {
		_, err := h.GetUser(context.Background(), tc.req)
		biz, ok := kerrors.FromBizStatusError(err)
		if !ok || biz.BizStatusCode() != tc.code {
			t.Fatalf("want %d got %v", tc.code, err)
		}
	}
}

如何验证业务异常经过真实 RPC 传回？​

internal/transport/kitex/rpc_test.go
package kitextransport

import (
	"context"
	"net"
	"testing"
	"time"

	domain "example.com/user-service/internal/user"
	"example.com/user-service/kitex_gen/user"
	"example.com/user-service/kitex_gen/user/userservice"
	"github.com/cloudwego/kitex/client"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"github.com/cloudwego/kitex/pkg/transmeta"
	"github.com/cloudwego/kitex/server"
	"github.com/cloudwego/kitex/transport"
)

func TestBizStatusOverTTHeader(t *testing.T) {
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	svc, _ := domain.NewService(missingRepo{})
	h, _ := NewUserServiceImpl(svc)
	exit := make(chan error, 1)
	svr := userservice.NewServer(h, server.WithListener(listener),
		server.WithMetaHandler(transmeta.ServerTTHeaderHandler),
		server.WithExitSignal(func() <-chan error { return exit }))
	done := make(chan error, 1)
	go func() { done <- svr.Run() }()
	defer func() {
		exit <- nil
		select {
		case err := <-done:
			if err != nil {
				t.Error(err)
			}
		case <-time.After(10 * time.Second):
			t.Error("server did not stop")
		}
	}()
	cli, err := userservice.NewClient("user-service",
		client.WithShortConnection(),
		client.WithHostPorts(listener.Addr().String()),
		client.WithRPCTimeout(time.Second),
		client.WithTransportProtocol(transport.TTHeader),
		client.WithMetaHandler(transmeta.ClientTTHeaderHandler))
	if err != nil {
		t.Fatal(err)
	}
	_, err = cli.GetUser(context.Background(), &user.GetUserRequest{Id: 1})
	biz, ok := kerrors.FromBizStatusError(err)
	if !ok || biz.BizStatusCode() != 40401 {
		t.Fatalf("business status lost over wire: %v", err)
	}
}
  • 范围: 临时 TCP 端口加生成 Client 验证 TTHeader 业务码；存储是 fake，测试用短连接便于退出;

如何执行编译与测试？​

go mod tidy
go test ./...
go vet ./...
go build ./...
# 有可用 C 编译器时检查并发问题
go test -race ./...
  • 保存: IDL、生成代码、go.mod 与 go.sum 一起保存；工具升级后重新生成并验证;

如何准备数据库与应用容器？​

compose.yaml
services:
  mysql:
    image: mysql:8.4
    environment:
      MYSQL_DATABASE: notes
      MYSQL_USER: notes
      MYSQL_PASSWORD: notes_dev
      MYSQL_ROOT_PASSWORD: root_dev
    ports: ["127.0.0.1:3306:3306"]
    volumes: ["db:/var/lib/mysql"]
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
      interval: 2s
      timeout: 2s
      retries: 30
  app:
    build: .
    environment:
      USER_DATABASE_DSN: "notes:notes_dev@tcp(mysql:3306)/notes?parseTime=true&loc=UTC"
    ports: ["127.0.0.1:8888:8888"]
    depends_on:
      mysql: { condition: service_healthy }
volumes:
  db:
Dockerfile
FROM golang:1.26 AS build
WORKDIR /src
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 go build -trimpath -o /out/server ./cmd/server
FROM gcr.io/distroless/static-debian12:nonroot
COPY --from=build /out/server /server
USER nonroot:nonroot
ENTRYPOINT ["/server"]
.dockerignore
.git
.env
*.exe
  • 本地: 示例密码只用于学习，端口绑定 127.0.0.1，两个项目不要同时占用相同端口;
  • 数据: 命名卷保存 MySQL 数据；应用以 nonroot 运行，真实发布按团队要求固定镜像 digest;

如何配置本地环境并初始化数据库？​

docker compose up -d --wait mysql
$env:USER_DATABASE_DSN = 'notes:notes_dev@tcp(127.0.0.1:3306)/notes?parseTime=true&loc=UTC'
go run ./cmd/migrate
  • 终端: 环境变量只属于当前终端；另开服务终端时必须设置相同 DSN，Hertz 还要复制同一个 JWT 密钥;

如何启动服务并验收接口？​

go run ./cmd/server
  • 请求终端: 服务保持运行，在另一个终端执行下面命令；Hertz 请求终端使用上节生成的 Token;
go run ./cmd/client
  • 预期: 打印同一用户的 ID、邮箱和姓名；改为固定邮箱重复创建可验证 40901;
  • 停止: 本机应用 Ctrl+C，检查在途请求结束后退出;

如何验证容器交付？​

docker compose up --build -d app
# 使用上一节相同请求验收
docker compose logs app
docker compose down
  • 前提: 已执行迁移，并停止占用 8888 的本机应用；Hertz 当前终端保留 ARTICLE_JWT_KEY;
  • 清理: down 保留命名卷，下一次启动可继续使用数据;

代码来源和验证范围是什么？​

  • 框架: CloudWeGo kitex 文档；生成、Handler、协议与生命周期依据官方资料;
  • 应用示例: 业务分层、数据模型、CRUD 与迁移命令由本笔记组织，不是框架强制架构;
  • 数据库: GORM 错误处理、字段更新;
  • 已验证: 代码生成、go test、go vet、go build；当前环境无 Docker 和 C 编译器，MySQL/Compose 与 race 检查需在具备依赖的环境执行;