如何用Go语言开发分布式系统的组件,涵盖服务注册发现、RPC通信、分布式锁等关键技术

// 典型Go协程示例
func worker(id int, jobs <-chan int, results chan<- int) {
    for j := range jobs {
        fmt.Printf("worker %d 开始处理任务 %d\n", id, j)
        time.Sleep(time.Second) // 模拟耗时操作
        results <- j * 2
    }
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    
    // 启动3个协程
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }
    
    // 发送5个任务
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)
    
    // 获取结果
    for a := 1; a <= 5; a++ {
        <-results
    }
}

// 服务注册示例（使用go.etcd.io/etcd/client/v3）
func registerService(serviceName, endpoint string) {
    client, _ := clientv3.New(clientv3.Config{
        Endpoints:   []string{"localhost:2379"},
        DialTimeout: 5 * time.Second,
    })

    // 创建租约（有效期10秒）
    resp, _ := client.Grant(context.TODO(), 10)
    
    // 注册服务节点
    key := fmt.Sprintf("/services/%s/%s", serviceName, endpoint)
    client.Put(context.TODO(), key, endpoint, clientv3.WithLease(resp.ID))

    // 保持心跳
    keepAlive, _ := client.KeepAlive(context.TODO(), resp.ID)
    go func() {
        for {
            <-keepAlive  // 每10秒自动续约
        }
    }()
}

// proto/order.proto
syntax = "proto3";
package order;

service OrderService {
    rpc CreateOrder (OrderRequest) returns (OrderResponse) {}
}

message OrderRequest {
    string product_id = 1;
    int32 quantity = 2;
}

message OrderResponse {
    string order_id = 1;
    int64 create_time = 2;
}

// gRPC服务端实现
type server struct {
    pb.UnimplementedOrderServiceServer
}

func (s *server) CreateOrder(ctx context.Context, in *pb.OrderRequest) (*pb.OrderResponse, error) {
    return &pb.OrderResponse{
        OrderId:    uuid.New().String(),
        CreateTime: time.Now().Unix(),
    }, nil
}

func startServer() {
    lis, _ := net.Listen("tcp", ":50051")
    s := grpc.NewServer()
    pb.RegisterOrderServiceServer(s, &server{})
    s.Serve(lis)
}

// 基于redis的分布式锁（使用go-redis/v9）
func acquireLock(client *redis.Client, lockKey string, timeout time.Duration) bool {
    // 尝试获取锁（NX表示不存在时设置）
    result, _ := client.SetNX(context.Background(), lockKey, "locked", timeout).Result()
    return result
}

func releaseLock(client *redis.Client, lockKey string) {
    // 使用Lua脚本保证原子性
    script := redis.NewScript(`
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end`)
    script.Run(context.Background(), client, []string{lockKey}, "locked")
}

type Server struct {
    Address string
    Weight  int
    current int
}

func nextServer(servers []*Server) *Server {
    total := 0
    var selected *Server
    
    for _, s := range servers {
        total += s.Weight
        s.current += s.Weight
        if selected == nil || s.current > selected.current {
            selected = s
        }
    }
    
    selected.current -= total
    return selected
}

// 使用示例
servers := []*Server{
    {"192.168.1.10", 3, 0},
    {"192.168.1.11", 2, 0},
    {"192.168.1.12", 1, 0},
}

for i := 0; i < 6; i++ {
    s := nextServer(servers)
    fmt.Println("选择服务器:", s.Address)
}