Golang rabbitMQ生产者消费者
Jeff的技术栈 人气:0消费者
package main import ( "fmt" "github.com/streadway/amqp" ) func failOnError(err error, msg string) { if err != nil { fmt.Println("%s: %s", msg, err) } } // 只能在安装 rabbitmq 的服务器上操作 func main() { conn, err := amqp.Dial("amqp://mquser:123456@127.0.0.1/toutiao") failOnError(err, "Failed to connect to RabbitMQ") defer conn.Close() ch, err := conn.Channel() failOnError(err, "Failed to open a channel") defer ch.Close() q, err := ch.QueueDeclare( "toutiao.web.test", // 队列名字 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 "toutiao", // consumer true, // auto-ack,true消费了就消失 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 { fmt.Println(fmt.Sprintf("返回的消息:%s",d.Body)) } }() fmt.Println("[*] Waiting for messages. To exit press CTRL+C") <-forever }
生产者
body:消息体
package main import ( "github.com/streadway/amqp" "log" ) func failOnError(err error, msg string) { if err != nil { log.Fatalf("%s: %s", msg, err) } } // 只能在安装 rabbitmq 的服务器上操作 func main() { conn, err := amqp.Dial("amqp://mquser:123456@127.0.0.1/toutiao") failOnError(err, "Failed to connect to RabbitMQ") defer conn.Close() ch, err := conn.Channel() failOnError(err, "Failed to open a channel") defer ch.Close() q, err := ch.QueueDeclare( "toutiao.web.test", // name false, // durable false, // delete when unused false, // exclusive false, // no-wait nil, // arguments ) failOnError(err, "Failed to declare a queue") body := "Hello World!" //发送的消息 err = 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(err, "Failed to publish a message") }
加载全部内容