package main import ( ) /* 默认点对点模式 工作方，多个，拿发布方的消息 */ func failOnError(err error, msg string) { if err != nil { log.Fatalf(, msg, err) panic(fmt.Sprintf(, msg, err)) } } func main() { conn, err := amqp.Dial() failOnError(err, ) defer conn.Close() ch, err := conn.Channel() failOnError(err, ) defer ch.Close() // 指定队列！ q, err := ch.QueueDeclare( , , , , , // no-wait nil, // arguments ) failOnError(err, ) // Fair dispatch 预取，每个工作方每次拿一个消息，确认后才拿下一次，缓解压力 err = ch.Qos( , , // global ) failOnError(err, ) // 消费根据队列名 msgs, err := ch.Consume( q.Name, , , , , , // no-wait nil, // args ) failOnError(err, ) forever := make(chan bool) go func() { for d := range msgs { log.Printf(, d.Body) dot_count := bytes.Count(d.Body, [])) t := time.Duration(dot_count) time.Sleep(t * time.Second) log.Printf() // 确认消息被收到！！如果为真的，那么同在一个channel,在该消息之前未确认的消息都会确认，适合批量处理 // 真时场景：每十条消息确认一次，类似 d.Ack(false) } }() log.Printf() <-forever }

package main import ( ) /* 广播模式 发布方 */ func failOnError(err error, msg string) { if err != nil { log.Fatalf(, msg, err) panic(fmt.Sprintf(, msg, err)) } } func main() { conn, err := amqp.Dial() failOnError(err, ) defer conn.Close() ch, err := conn.Channel() failOnError(err, ) defer ch.Close() // 默认模式有默认交换机，广播自己定义一个交换机，交换机可与队列进行绑定 err = ch.ExchangeDeclare( , , , , , , // no-wait nil, // arguments ) failOnError(err, ) body := bodyFrom(os.Args) // 发布 err = ch.Publish( , , , , // immediate amqp.Publishing{ ContentType: , Body: []byte(body), }) failOnError(err, ) log.Printf(, body) } func bodyFrom(args []string) string { var s string if (len(args) < 2) || os.Args[1] == "" { s = fmt.Sprintf(,, time.Now()) } else { s = strings.Join(args[1:], " ") } return s }

package main import ( ) /* 广播模式 订阅方 */ func failOnError(err error, msg string) { if err != nil { log.Fatalf(, msg, err) panic(fmt.Sprintf(, msg, err)) } } func main() { conn, err := amqp.Dial() failOnError(err, ) defer conn.Close() ch, err := conn.Channel() failOnError(err, ) defer ch.Close() // 同样要申明交换机 err = ch.ExchangeDeclare( , , , , , , // no-wait nil, // arguments ) failOnError(err, ) // 新建队列，这个队列没名字，随机生成一个名字 q, err := ch.QueueDeclare( , , , , // no-wait nil, // arguments ) failOnError(err, ) // 队列和交换机绑定，即是队列订阅了发到这个交换机的消息 err = ch.QueueBind( q.Name, , , , nil) failOnError(err, ) // 开始消费消息，可开多个订阅方，因为队列是临时生成的，所有每个订阅方都能收到同样的消息 msgs, err := ch.Consume( q.Name, , , , , , // no-wait nil, // args ) failOnError(err, ) forever := make(chan bool) go func() { for d := range msgs { log.Printf(, d.Body) } }() log.Printf() <-forever }