go实现消息队列

亦凉 2022-10-28 10:11 213阅读 0赞

## 一、rabbitmq ##

package main
    
    import (
    	"fmt"
    	"log"
    
    	"github.com/streadway/amqp"
    )
    
    func failOnError(err error, msg string) { 
    	if err != nil { 
    		log.Fatalf("%s: %s", msg, err)
    		panic(fmt.Sprintf("%s: %s", msg, err))
    	}
    }
    
    func main() { 
    	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    	failOnError(err, "Failed to connect to RabbitMQ")
    	defer conn.Close()
    
    	ch, err := conn.Channel()
    	failOnError(err, "Failed to open a channel")
    	defer ch.Close()
    
    	failOnError(err, "Failed to declare a queue")
    
    	err = ch.Qos(
    		1,     // prefetch count
    		0,     // prefetch size
    		false, //global
    	)
    	failOnError(err, "Failed to set Qos")
    
    	//发送消息
    	funcSend(ch)
    	//收消息
    	funcRecv(ch)
    
    }
    
    //生产者
    func funcSend(ch *amqp.Channel) { 
    	q, err := ch.QueueDeclare(
    		"queue", //name
    		false,   //durables
    		false,   //delete when unused
    		false,   //exclusive
    		false,   //no wait
    		nil,     //args
    	)
    	failOnError(err, "Failed to declare a queue")
    	body := "Hello World!"
    	error := ch.Publish(
    		"",     // exchange
    		q.Name, // routing key
    		false,  // mandatory
    		false,  // immediate
    		amqp.Publishing{ 
    			ContentType: "text/plain",
    			Body:        []byte(body),
    		})
    	log.Printf("[x] Sent %s", body)
    	failOnError(error, "Failed to publish a message")
    }
    
    //消费者
    func funcRecv(ch *amqp.Channel) { 
    	q, err := ch.QueueDeclare(
    		"queue", // name
    		false,   // durable
    		false,   // delete when unused
    		false,   // exclusive
    		false,   // no-wait
    		nil,     // arguments
    	)
    	failOnError(err, "Failed to declare a queue")
    	msgs, err := ch.Consume(
    		q.Name, // queue
    		"",     // consumer
    		true,   // auto-ack
    		false,  // exclusive
    		false,  // no-local
    		false,  // no-wait
    		nil,    // args
    	)
    	failOnError(err, "Failed to register a consumer")
    	forever := make(chan bool)
    	go func() { 
    		for d := range msgs { 
    			log.Printf("Received a message: %s", d.Body)
    		}
    	}()
    	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
    	<-forever
    }

## 二、nats-io ##

package main
    
    import (
    	"flag"
    	"log"
    	"strconv"
    	"time"
    
    	"github.com/nats-io/nats.go"
    	"github.com/pborman/uuid"
    )
    
    const (
    	//url = "nats://192.168.3.125:4222"
    	url = nats.DefaultURL
    )
    
    var (
    	nc  *nats.Conn
    	err error
    )
    
    func init() { 
    	// if nc, err = nats.Connect(url); checkErr(err) { 
    	// fmt.
    	// }
    }
    
    func main() { 
    
    }
    
    //发送消息
    func sendMessage() { 
    	var (
    		subj = flag.String("subj", "", "subject name")
    	)
    	flag.Parse()
    	log.Println(*subj)
    	startClient(*subj)
    	time.Sleep(time.Second)
    }
    
    func startClient(subj string) { 
    	for i := 0; i < 1; i++ { 
    		id := uuid.New()
    		log.Println(id)
    		nc.Publish(subj, []byte(id+" Sun "+strconv.Itoa(i)))
    		nc.Publish(subj, []byte(id+" Rain "+strconv.Itoa(i)))
    		nc.Publish(subj, []byte(id+" Fog "+strconv.Itoa(i)))
    		nc.Publish(subj, []byte(id+" Cloudy "+strconv.Itoa(i)))
    	}
    }
    
    //接收消息
    func receiveMessage() { 
    	var (
    		servername = flag.String("servername", "y", "name for server")
    		queueGroup = flag.String("group", "", "group name for Subscribe")
    		subj       = flag.String("subj", "", "subject name")
    	)
    	flag.Parse()
    
    	log.Println(*servername, *queueGroup, *subj)
    	startService(*subj, *servername+" worker1", *queueGroup)
    	startService(*subj, *servername+" worker2", *queueGroup)
    	startService(*subj, *servername+" worker3", *queueGroup)
    	select { }
    }
    
    func startService(subj, name, queue string) { 
    	go async(nc, subj, name, queue)
    }
    
    func async(nc *nats.Conn, subj, name, queue string) { 
    	nc.QueueSubscribe(subj, queue, func(msg *nats.Msg) { 
    		log.Println(name, "Received a message From Async : ", string(msg.Data))
    	})
    }
    
    func checkErr(err error) bool { 
    	if err != nil { 
    		log.Println(err)
    		return false
    	}
    	return true
    }