golang redigo订阅分布，stream-Go服务器开发

Redigo是Redis的一个Golang客户端，支持订阅/发布和Stream数据类型。在这里，我们将讨论如何使用redigo来实现分布式系统中的消息订阅和流处理。

订阅/发布

在Redis中，可以使用SUBSCRIBE命令订阅通道，并使用PUBLISH命令向通道发送消息。redigo为这些操作提供了方便的API。

首先，需要创建一个连接到Redis服务器的连接池：

pool := &redis.Pool{
    MaxIdle:     3,
    IdleTimeout: 240 * time.Second,
    Dial: func() (redis.Conn, error) {
        c, err := redis.Dial("tcp", ":6379")
        if err != nil {
            return nil, err
        }
        return c, err
    },
}
defer pool.Close()

接下来，可以使用以下代码订阅通道：

conn := pool.Get()
defer conn.Close()

psc := redis.PubSubConn{Conn: conn}

// 订阅通道
psc.Subscribe("channel")

// 处理消息
for {
    switch v := psc.Receive().(type) {
    case redis.Message:
        fmt.Printf("%s: message: %s\n", v.Channel, v.Data)
    case redis.Subscription:
        fmt.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)
    case error:
        // 处理错误
        return
    }
}

上面的代码通过创建PubSubConn对象来进行订阅。然后，在无限循环中等待接收消息。当有新消息到达时，它将被打印出来。如果订阅发生错误，则循环将退出。

发送消息的方法与普通命令一样简单：

conn := pool.Get()
defer conn.Close()

_, err := conn.Do("PUBLISH", "channel", "hello world")
if err != nil {
    // 处理错误
}

Stream

Stream是Redis 5.0引入的数据类型，可以用于实现高性能的流处理应用程序。redigo为Stream提供了方便的API。

首先，需要创建一个连接到Redis服务器的连接池：

pool := &redis.Pool{
    MaxIdle:     3,
    IdleTimeout: 240 * time.Second,
    Dial: func() (redis.Conn, error) {
        c, err := redis.Dial("tcp", ":6379")
        if err != nil {
            return nil, err
        }
        return c, err
    },
}
defer pool.Close()

接下来，可以使用以下代码向Stream添加条目：

conn := pool.Get()
defer conn.Close()

// 添加条目到Stream中
_, err := conn.Do("XADD", "mystream", "*", "message", "hello world")
if err != nil {
    // 处理错误
}

上面的代码通过执行XADD命令将一条包含“hello world”消息的新条目添加到名为“mystream”的Stream中。

可以使用以下代码从流中读取最后几个元素：

conn := pool.Get()
defer conn.Close()

// 从流中读取最后两个元素
res, err := redis.Values(conn.Do("XREVRANGE", "mystream", "+", "-", "COUNT", 2))
if err != nil {
    // 处理错误
}

// 解析结果
for len(res) > 0 {
    var id string
    var values []interface{}
    res, _ = redis.Scan(res, &id, &values)

    fmt.Printf("%s: %v\n", id, values)
}

上面的代码通过执行XREVRANGE命令从名为“mystream”的Stream中读取最后两个元素。然后，使用redis.Scan解析结果，并将它们打印出来。

总之，redigo提供了方便的API来处理订阅/发布和Stream数据类型。这些功能使得它成为构建分布式系统和流处理应用程序的强大工具。