golang框架在分布式系统中的应用实例

package queue

import (
    "context"
    "fmt"
    "github.com/fasthttp/websocket"
    "github.com/streadway/amqp"
    "log"
    "sync"
)

// 任务队列
type Queue struct {
    tasks chan []byte
    mu    sync.Mutex
}

// 创建新的任务队列
func NewQueue() *Queue {
    return &Queue{
        tasks: make(chan []byte),
    }
}

// 添加任务到队列
func (q *Queue) AddTask(data []byte) {
    q.mu.Lock()
    defer q.mu.Unlock()
    q.tasks <- data
}

// 启动队列服务
func (q *Queue) Start(ctx context.Context) error {
    // 连接到 RabbitMQ，创建一个发布者和消费者。
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        return err
    }
    pubsub, err := conn.Channel()
    if err != nil {
        return err
    }
    defer pubsub.Close()

    // 订阅一个匿名的队列，并接收消息。
    queue, err := pubsub.QueueDeclare("", false, false, false, false, nil)
    if err != nil {
        return err
    }

    msgs, err := pubsub.Consume(
        queue.Name,
        "",
        false,
        false,
        false,
        false,
        nil,
    )
    if err != nil {
        return err
    }

    // 处理来自客户端的 WebSocket 请求。
    websocket.WebSocketHandler(func(conn *websocket.Conn) {

        // 从队列中取任务并将其传递给客户端。
        go func() {
            for task := range q.tasks {
                if err := conn.WriteMessage(websocket.MessageBinary, task); err != nil {
                    log.Printf("WebSocket 写入失败：%v", err)
                    break
                }
            }
            conn.Close()
        }()

        // 从队列中接收 WebSocket 消息。
        for {
            messageType, message, err := conn.ReadMessage()
            if err != nil {
                log.Printf("WebSocket 读取失败：%v", err)
                break
            }
            if messageType == websocket.CloseMessage {
                break
            }
            // 处理客户端发送的消息。
            q.handleMessage(message)
        }
        conn.Close()
    }).ServeHTTP(&fasthttp.Server{})
    return nil
}

// 处理客户端发送的消息
func (q *Queue) handleMessage(data []byte) {
    // 业务逻辑在此处实现，例如处理任务。
    // ...
}

package worker

import (
    "context"
    "fmt"
    "github.com/streadway/amqp"
    "log"
    "os"
    "sync"
    "time"
)

// 工作节点
type Worker struct {
    conn    *amqp.Connection
    channel *amqp.Channel
    queue   amqp.Queue
    tasks   chan amqp.Delivery
    wg      sync.WaitGroup
}

// 创建新的工作者
func NewWorker(ctx context.Context, amqpURL, queueName string) (*Worker, error) {
    conn, err := amqp.Dial(amqpURL)
    if err != nil {
        return nil, err
    }

    channel, err := conn.Channel()
    if err != nil {
        return nil, err
    }

    // 声明队列，如果队列不存在则创建。
    queue, err := channel.QueueDeclare(
        queueName, // 队列名称
        false,     // 持久性
        false,     // 独占
        false,     // 删除未使用队列
        false,     // 等待接收者
        nil,       // 其他参数
    )
    if err != nil {
        return nil, err
    }

    return &Worker{
        conn:    conn,
        channel: channel,
        queue:   queue,
        tasks:   make(chan amqp.Delivery),
    }, nil
}

// 开始处理任务
func (w *Worker) Start(ctx context.Context) error {
    w.wg.Add(1)

    go func() {
        defer w.wg.Done()
        for {
            delivery, ok := <-w.tasks

            // 处理任务。
            // 业务逻辑在此处实现。
            // ...

            // 将确认发送给 RabbitMQ，表示该任务已完成。
            if delivery.Acknowledger != nil {
                if err := delivery.Ack(false); err != nil {
                    log.Fatalf("无法确认任务：%v", err)
                }
            }
        }
    }()

    // 从队列中接收任务
    log.Printf("工作者 %s 正在监听队列 %s...", os.Args[0], w.queue.Name)
    msgs, err := w.channel.Consume(
        w.queue.Name,
        "",
        false, // 自动确认
        false, // 仅消费一个消费者
        false, // 排他
        false, // 不等待响应