Golang实现RabbitMQ消费者的定时任务

在Go语言中实现RabbitMQ消费者时，不应该使用定时器轮询的方式。RabbitMQ的消费者设计应该是持续监听队列，而不是定时检查。以下是修正后的实现：

package main

import (
	"encoding/json"
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"os"
	"os/signal"
	"syscall"
)

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	HandleError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	HandleError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare(
		"hello",
		false,
		false,
		false,
		false,
		nil,
	)
	HandleError(err, "could not declare 'hello' queue")

	err = ch.Qos(1, 0, false)
	HandleError(err, "could not configure Qos")

	msgs, err := ch.Consume(
		q.Name,
		"",
		false, // 设置为false，手动确认消息
		false,
		false,
		false,
		nil,
	)
	HandleError(err, "could not register consumer")

	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

	log.Printf("Consumer ready, PID: %d", os.Getpid())
	log.Printf("[*] waiting for messages. To exit press CTRL+C")

	for {
		select {
		case d := <-msgs:
			log.Printf("Received a message: %s", string(d.Body))
			locationData := &RabbitLocationData{}

			err := json.Unmarshal(d.Body, locationData)
			if err != nil {
				log.Printf("Error decoding JSON: %s", err)
				d.Nack(false, false) // 解码失败，拒绝消息
				continue
			}

			log.Printf("Result: %s", locationData.ToString())

			// 这里添加数据库插入逻辑
			// err = insertToDatabase(locationData)
			// if err != nil {
			//     log.Printf("Error inserting to database: %s", err)
			//     d.Nack(false, true) // 数据库失败，重新入队
			//     continue
			// }

			if err := d.Ack(false); err != nil {
				log.Printf("Error acknowledging message: %s", err)
			} else {
				log.Printf("Acknowledged message")
			}

		case <-sigs:
			log.Println("Received termination signal, shutting down...")
			return
		}
	}
}

func HandleError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

type RabbitLocationData struct {
	Domain             string  `json:"domain"`
	UserId             string  `json:"user_id"`
	ClientTimeStampUTC float64 `json:"client_timestamp_utc"`
	ServerTimeStampUTC float64 `json:"server_timestamp_utc"`
	Longitude          float64 `json:"lon"`
	Latitude           float64 `json:"lat"`
}

func (rld RabbitLocationData) ToString() string {
	result := fmt.Sprintf("RabbitLocationData{userId='%s', "+
		"clientTimestampUtc=%f, "+
		"serverTimestampUtc=%f, "+
		"latitude=%f, "+
		"longitude=%f, "+
		"domain='%s'"+
		"}",
		rld.UserId,
		rld.ClientTimeStampUTC,
		rld.ServerTimeStampUTC,
		rld.Latitude,
		rld.Longitude,
		rld.Domain)
	return result
}

如果需要并发处理消息，可以使用goroutine池：

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	HandleError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	HandleError(err, "Failed to open a channel")
	defer ch.Close()

	q, err := ch.QueueDeclare(
		"hello",
		false,
		false,
		false,
		false,
		nil,
	)
	HandleError(err, "could not declare 'hello' queue")

	err = ch.Qos(10, 0, false) // 增加预取数量
	HandleError(err, "could not configure Qos")

	msgs, err := ch.Consume(
		q.Name,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
	HandleError(err, "could not register consumer")

	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

	workerCount := 5
	done := make(chan bool)

	for i := 0; i < workerCount; i++ {
		go func(workerID int) {
			for d := range msgs {
				log.Printf("Worker %d received message: %s", workerID, string(d.Body))
				
				locationData := &RabbitLocationData{}
				err := json.Unmarshal(d.Body, locationData)
				
				if err != nil {
					log.Printf("Worker %d error decoding JSON: %s", workerID, err)
					d.Nack(false, false)
					continue
				}

				log.Printf("Worker %d processing: %s", workerID, locationData.ToString())
				
				// 数据库插入逻辑
				// if err := insertToDatabase(locationData); err != nil {
				//     d.Nack(false, true)
				//     continue
				// }
				
				d.Ack(false)
			}
			done <- true
		}(i)
	}

	log.Printf("Started %d workers, waiting for messages...", workerCount)

	<-sigs
	log.Println("Shutting down workers...")
	ch.Close()
	
	for i := 0; i < workerCount; i++ {
		<-done
	}
}

这个实现使用了RabbitMQ的原生消费模式，消息到达时会立即被处理，不需要定时轮询。Consume方法返回的channel会持续接收消息，直到连接关闭。