Go语言中实现可靠后台任务处理的策略与实践

package main

import (
    "fmt"
    "net/http"
    "time"
)

func sendEmail(to, subject, body string) {
    fmt.Printf("Sending email to %s: Subject '%s'\n", to, subject)
    time.Sleep(5 * time.Second) // Simulate network delay and processing
    fmt.Printf("Email sent to %s\n", to)
}

func signupHandler(w http.ResponseWriter, r *http.Request) {
    userEmail := r.FormValue("email")
    if userEmail == "" {
        http.Error(w, "Email is required", http.StatusBadRequest)
        return
    }

    // 模拟用户注册逻辑
    fmt.Printf("User %s registered successfully.\n", userEmail)

    // 启动Goroutine异步发送邮件
    go sendEmail(userEmail, "Welcome to our service!", "Thank you for registering.")

    w.WriteHeader(http.StatusOK)
    w.Write([]byte("Registration successful! Confirmation email will be sent shortly."))
}

func main() {
    http.HandleFunc("/signup", signupHandler)
    fmt.Println("Server listening on :8080")
    http.ListenAndServe(":8080", nil)
}

package main

import (
    "encoding/json"
    "fmt"
    "log"
    "time"

    // 假设这里引入了某个队列服务的客户端库，例如：
    // "github.com/your-queue-client"
)

// Task represents a background job
type Task struct {
    Type    string                 `json:"type"`
    Payload map[string]interface{} `json:"payload"`
}

// PushTaskToQueue simulates pushing a task to a distributed queue
func PushTaskToQueue(task Task) error {
    taskBytes, err := json.Marshal(task)
    if err != nil {
        return fmt.Errorf("failed to marshal task: %w", err)
    }

    // In a real application, you would connect to RabbitMQ, Beanstalkd, or Redis
    // and publish/push taskBytes to a specific queue.
    // For demonstration, we just print it.
    fmt.Printf("[%s] Producer: Pushing task to queue: %s\n", time.Now().Format("15:04:05"), string(taskBytes))

    // Example with a hypothetical queue client:
    // client, err := yourqueueclient.NewClient("amqp://guest:guest@localhost:5672/")
    // if err != nil {
    //     return fmt.Errorf("failed to connect to queue: %w", err)
    // }
    // defer client.Close()
    //
    // err = client.Publish("email_queue", taskBytes)
    // if err != nil {
    //     return fmt.Errorf("failed to publish task: %w", err)
    // }

    return nil
}

func main() {
    // Simulate a user signup event triggering an email task
    emailTask := Task{
        Type: "send_confirmation_email",
        Payload: map[string]interface{}{
            "to":      "user@example.com",
            "subject": "Welcome!",
            "body":    "Thank you for registering!",
        },
    }

    if err := PushTaskToQueue(emailTask); err != nil {
        log.Fatalf("Error pushing email task: %v", err)
    }

    fmt.Println("Producer finished. Task sent to queue.")
    // In a real web server, this would be part of an HTTP handler.
    // The main goroutine would continue serving requests.
}

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    // 假设这里引入了某个队列服务的客户端库
    // "github.com/your-queue-client"
)

// Task represents a background job (same as in producer)
type Task struct {
    Type    string                 `json:"type"`
    Payload map[string]interface{} `json:"payload"`
}

// processEmailTask simulates sending an email
func processEmailTask(payload map[string]interface{}) error {
    to := payload["to"].(string)
    subject := payload["subject"].(string)
    body := payload["body"].(string)

    fmt.Printf("[%s] Worker: Processing email to %s (Subject: %s)\n", time.Now().Format("15:04:05"), to, subject)
    time.Sleep(3 * time.Second) // Simulate email sending delay
    // Simulate a potential failure for demonstration
    if time.Now().Second()%2 == 0 { // Every other time, simulate failure
        return fmt.Errorf("simulated email sending failure to %s", to)
    }
    fmt.Printf("[%s] Worker: Email successfully sent to %s\n", time.Now().Format("15:04:05"), to)
    return nil
}

// StartWorker simulates a worker pulling tasks from a distributed queue
func StartWorker(ctx context.Context) {
    fmt.Println("Worker started. Waiting for tasks...")

    // In a real application, you would connect to RabbitMQ, Beanstalkd, or Redis
    // and start consuming messages from a specific queue.
    // For demonstration, we simulate receiving tasks.

    // Example with a hypothetical queue client:
    // client, err := yourqueueclient.NewClient("amqp://guest:guest@localhost:5672/")
    // if err != nil {
    //     log.Fatalf("Failed to connect to queue: %v", err)
    // }
    // defer client.Close()
    //
    // messages, err := client.Consume("email_queue")
    // if err != nil {
    //     log.Fatalf("Failed to register consumer: %v", err)
    // }

    // Simulate receiving messages
    simulatedQueue := make(chan []byte, 10)
    go func() {
        // This goroutine simulates tasks being added to the queue over time
        for i := 0; ; i++ {
            select {
            case <-ctx.Done():
                return
            case simulatedQueue <- []byte(fmt.Sprintf(`{"type":"send_confirmation_email","payload":{"to":"user%d@example.com","subject":"Welcome %d!","body":"Thank you for registering!"}}`, i, i)):
                time.Sleep(1 * time.Second) // Simulate tasks arriving
            }
        }
    }()

    for {
        select {
        case <-ctx.Done():
            fmt.Println("Worker received shutdown signal, stopping...")
            return
        case msgBytes := <-simulatedQueue: // In real app: msgBytes := <-messages
            var task Task
            if err := json.Unmarshal(msgBytes, &task); err != nil {
                log.Printf("Worker: Failed to unmarshal task: %v, message: %s", err, string(msgBytes))
                // In a real system, you might send this to a dead-letter queue
                continue
            }

            fmt.Printf("[%s] Worker: Received task type: %s\n", time.Now().Format("15:04:05"), task.Type)

            var processingErr error
            switch task.Type {
            case "send_confirmation_email":
                processingErr = processEmailTask(task.Payload)
            default:
                log.Printf("Worker: Unknown task type: %s", task.Type)
            }

            if processingErr != nil {
                log.Printf("[%s] Worker: Task processing failed for type %s: %v", time.Now().Format("15:04:05"), task.Type, processingErr)
                // In a real system:
                // If using RabbitMQ, Nack the message with re-queue=true or send to dead-letter queue.
                // If using Beanstalkd, Bury the job or Release it with a delay.
            } else {
                // In a real system:
                // Acknowledge the message to the queue to remove it.
                fmt.Printf("[%s] Worker: Task type %s completed successfully.\n", time.Now().Format("15:04:05"), task.Type)
            }
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Handle graceful shutdown signals
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    go StartWorker(ctx)

    <-sigChan // Block until a signal is received
    fmt.Println("\nMain: Shutting down gracefully...")
    cancel() // Signal worker to stop
    time.Sleep(2 * time.Second) // Give worker some time to clean up
    fmt.Println("Main: Shutdown complete.")
}