在Beego中使用kafka实现消息队列

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

package main

import (
    "github.com/Shopify/sarama"
)

func main() {
    // 设置kafka配置
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true

    // 新建生产者
    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
    if err != nil {
        panic(err)
    }

    // 构造消息
    message := &sarama.ProducerMessage{
        Topic: "test",
        Value: sarama.StringEncoder("test message"),
    }

    // 发送消息
    _, _, err = producer.SendMessage(message)
    if err != nil {
        panic(err)
    }

    producer.Close()
}

package main

import (
    "fmt"
    "github.com/Shopify/sarama"
    "log"
    "os"
    "os/signal"
)

func main() {
    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true

    // 新建一个消费者
    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
    if err != nil {
        panic(err)
    }

    // 准备订阅话题
    topic := "test"
    partitionList, err := consumer.Partitions(topic)
    if err != nil {
        panic(err)
    }

    // 启动goroutine处理消息
    for _, partition := range partitionList {
        // 构造一个partitionConsumer
        pc, err := consumer.ConsumePartition(topic, partition, sarama.OffsetNewest)
        if err != nil {
            panic(err)
        }

        go func(partitionConsumer sarama.PartitionConsumer) {
            defer func() {
                // 关闭consumer
                if err := partitionConsumer.Close(); err != nil {
                    log.Fatalln(err)
                }
            }()
            for msg := range partitionConsumer.Messages() {
                fmt.Printf("Partition:%d Offset:%d Key:%s Value:%s
",
                    msg.Partition, msg.Offset, msg.Key, msg.Value)
            }
        }(pc)
    }

    // 处理中断信号
    sigterm := make(chan os.Signal, 1)
    signal.Notify(sigterm, os.Interrupt)
    <-sigterm

    fmt.Println("Shutdown")
    consumer.Close()
}