编译工程生产消费

引入依赖

go.mod

module test_go

require (
github.com/rabbitmq/amqp091-go v1.10.0
golang.org/x/net v0.26.0
)

go 1.20

生产消息

package main

import (
"flag"
"fmt"
amqp "github.com/rabbitmq/amqp091-go"
"log"
)

var (
uri          = flag.String("uri", "amqp://USERNAME:PASSWORD@33.0.1.35:5672", "AMQP URI")
exchangeName = flag.String("exchange", "test-exchange", "Durable AMQP exchange name")
exchangeType = flag.String("exchange-type", "direct", "Exchange type - direct|fanout|topic|x-custom")
routingKey   = flag.String("key", "test-key", "AMQP routing key")
body         = flag.String("body", "foobar", "Body of message")
reliable     = flag.Bool("reliable", true, "Wait for the publisher confirmation before exiting")
)

func init() {
flag.Parse()
}

func main() {
if err := publish(*uri, *exchangeName, *exchangeType, *routingKey, *body, *reliable); err != nil {
log.Fatalf("%s", err)
}
log.Printf("published %dB OK", len(*body))
}

func publish(amqpURI, exchange, exchangeType, routingKey, body string, reliable bool) error {

// This function dials, connects, declares, publishes, and tears down,
// all in one go. In a real service, you probably want to maintain a
// long-lived connection as state, and publish against that.

log.Printf("dialing %q", amqpURI)
connection, err := amqp.Dial(amqpURI)
if err != nil {
return fmt.Errorf("Dial: %s", err)
}
defer connection.Close()

log.Printf("got Connection, getting Channel")
channel, err := connection.Channel()
if err != nil {
return fmt.Errorf("Channel: %s", err)
}

log.Printf("got Channel, declaring %q Exchange (%q)", exchangeType, exchange)
if err := channel.ExchangeDeclare(
exchange,     // name
exchangeType, // type
true,         // durable
false,        // auto-deleted
false,        // internal
false,        // noWait
nil,          // arguments
); err != nil {
return fmt.Errorf("Exchange Declare: %s", err)
}

// Reliable publisher confirms require confirm.select support from the
// connection.
if reliable {
log.Printf("enabling publishing confirms.")
if err := channel.Confirm(false); err != nil {
return fmt.Errorf("Channel could not be put into confirm mode: %s", err)
}

confirms := channel.NotifyPublish(make(chan amqp.Confirmation, 1))

defer confirmOne(confirms)
}

log.Printf("declared Exchange, publishing %dB body (%q)", len(body), body)
if err = channel.Publish(
exchange,   // publish to an exchange
routingKey, // routing to 0 or more queues
false,      // mandatory
false,      // immediate
amqp.Publishing{
Headers:         amqp.Table{},
ContentType:     "text/plain",
ContentEncoding: "",
Body:            []byte(body),
DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
Priority:        0,              // 0-9
// a bunch of application/implementation-specific fields
},
); err != nil {
return fmt.Errorf("Exchange Publish: %s", err)
}

return nil
}

// One would typically keep a channel of publishings, a sequence number, and a
// set of unacknowledged sequence numbers and loop until the publishing channel
// is closed.
func confirmOne(confirms <-chan amqp.Confirmation) {
log.Printf("waiting for confirmation of one publishing")

if confirmed := <-confirms; confirmed.Ack {
log.Printf("confirmed delivery with delivery tag: %d", confirmed.DeliveryTag)
} else {
log.Printf("failed delivery of delivery tag: %d", confirmed.DeliveryTag)
}
}

消费消息

package main

import (
"flag"
"fmt"
amqp "github.com/rabbitmq/amqp091-go"
"log"
"time"
)

var (
uri          = flag.String("uri", "amqp://USERNAME:PASSWORD@10.10.33.196:5672", "AMQP URI")
exchange     = flag.String("exchange", "test-exchange", "Durable, non-auto-deleted AMQP exchange name")
exchangeType = flag.String("exchange-type", "direct", "Exchange type - direct|fanout|topic|x-custom")
queue        = flag.String("queue", "test-queue", "Ephemeral AMQP queue name")
bindingKey   = flag.String("key", "test-key", "AMQP binding key")
consumerTag  = flag.String("consumer-tag", "simple-consumer", "AMQP consumer tag (should not be blank)")
lifetime     = flag.Duration("lifetime", 5*time.Second, "lifetime of process before shutdown (0s=infinite)")
)

func init() {
flag.Parse()
}

func main() {
c, err := NewConsumer(*uri, *exchange, *exchangeType, *queue, *bindingKey, *consumerTag)
if err != nil {
log.Fatalf("%s", err)
}

if *lifetime > 0 {
log.Printf("running for %s", *lifetime)
time.Sleep(*lifetime)
} else {
log.Printf("running forever")
select {}
}

log.Printf("shutting down")

if err := c.Shutdown(); err != nil {
log.Fatalf("error during shutdown: %s", err)
}
}

type Consumer struct {
conn    *amqp.Connection
channel *amqp.Channel
tag     string
done    chan error
}

func NewConsumer(amqpURI, exchange, exchangeType, queueName, key, ctag string) (*Consumer, error) {
c := &Consumer{
conn:    nil,
channel: nil,
tag:     ctag,
done:    make(chan error),
}

var err error

log.Printf("dialing %q", amqpURI)
c.conn, err = amqp.Dial(amqpURI)
if err != nil {
return nil, fmt.Errorf("Dial: %s", err)
}

go func() {
fmt.Printf("closing: %s", <-c.conn.NotifyClose(make(chan *amqp.Error)))
}()

log.Printf("got Connection, getting Channel")
c.channel, err = c.conn.Channel()
if err != nil {
return nil, fmt.Errorf("Channel: %s", err)
}

log.Printf("got Channel, declaring Exchange (%q)", exchange)
if err = c.channel.ExchangeDeclare(
exchange,     // name of the exchange
exchangeType, // type
true,         // durable
false,        // delete when complete
false,        // internal
false,        // noWait
nil,          // arguments
); err != nil {
return nil, fmt.Errorf("Exchange Declare: %s", err)
}

log.Printf("declared Exchange, declaring Queue %q", queueName)
queue, err := c.channel.QueueDeclare(
queueName, // name of the queue
true,      // durable
false,     // delete when unused
false,     // exclusive
false,     // noWait
nil,       // arguments
)
if err != nil {
return nil, fmt.Errorf("Queue Declare: %s", err)
}

log.Printf("declared Queue (%q %d messages, %d consumers), binding to Exchange (key %q)",
queue.Name, queue.Messages, queue.Consumers, key)

if err = c.channel.QueueBind(
queue.Name, // name of the queue
key,        // bindingKey
exchange,   // sourceExchange
false,      // noWait
nil,        // arguments
); err != nil {
return nil, fmt.Errorf("Queue Bind: %s", err)
}

log.Printf("Queue bound to Exchange, starting Consume (consumer tag %q)", c.tag)
deliveries, err := c.channel.Consume(
queue.Name, // name
c.tag,      // consumerTag,
false,      // noAck
false,      // exclusive
false,      // noLocal
false,      // noWait
nil,        // arguments
)
if err != nil {
return nil, fmt.Errorf("Queue Consume: %s", err)
}

go handle(deliveries, c.done)

return c, nil
}

func (c *Consumer) Shutdown() error {
// will close() the deliveries channel
if err := c.channel.Cancel(c.tag, true); err != nil {
return fmt.Errorf("Consumer cancel failed: %s", err)
}

if err := c.conn.Close(); err != nil {
return fmt.Errorf("AMQP connection close error: %s", err)
}

defer log.Printf("AMQP shutdown OK")

// wait for handle() to exit
return <-c.done
}

func handle(deliveries <-chan amqp.Delivery, done chan error) {
for d := range deliveries {
log.Printf(
"got %dB delivery: [%v] %q",
len(d.Body),
d.DeliveryTag,
d.Body,
)
d.Ack(false)
}
log.Printf("handle: deliveries channel closed")
done <- nil
}

ssl生产消息

package main

import (
"crypto/tls"
"crypto/x509"
"flag"
"fmt"
amqp "github.com/rabbitmq/amqp091-go"
"io/ioutil"
"log"
)

var (
uri          = flag.String("uri", "amqps://USERNAME:PASSWORD@10.10.33.196:5671", "AMQP URI")
exchangeName = flag.String("exchange", "go-exchange", "Durable AMQP exchange name")
exchangeType = flag.String("exchange-type", "direct", "Exchange type - direct|fanout|topic|x-custom")
routingKey   = flag.String("key", "test-key", "AMQP routing key")
body         = flag.String("body", "foobar", "Body of message")
reliable     = flag.Bool("reliable", true, "Wait for the publisher confirmation before exiting")
)

func init() {
flag.Parse()
}

func main() {
if err := publish(*uri, *exchangeName, *exchangeType, *routingKey, *body, *reliable); err != nil {
log.Fatalf("%s", err)
}
log.Printf("published %dB OK", len(*body))
}

func publish(amqpsURI, exchange, exchangeType, routingKey, body string, reliable bool) error {
caCert, err := ioutil.ReadFile("D:\\tmp\\hzmq-test-0520_rabbitmq_ssl_client\\ca_certificate.pem")
if err != nil {
return err
}

cert, err := tls.LoadX509KeyPair("D:\\tmp\\hzmq-test-0520_rabbitmq_ssl_client\\client_rabbitmq_certificate.pem", "D:\\tmp\\hzmq-test-0520_rabbitmq_ssl_client\\client_rabbitmq_key.pem")
if err != nil {
return err
}

rootCAs := x509.NewCertPool()
rootCAs.AppendCertsFromPEM(caCert)

tlsConf := &tls.Config{
RootCAs:      rootCAs,
Certificates: []tls.Certificate{cert},
//ServerName:   "localhost", // Optional
InsecureSkipVerify: true,
}

connection, err := amqp.DialTLS(amqpsURI, tlsConf)
if err != nil {
return fmt.Errorf("Dial: %s", err)
}
defer connection.Close()

log.Printf("got Connection, getting Channel")
channel, err := connection.Channel()
if err != nil {
return fmt.Errorf("Channel: %s", err)
}

log.Printf("got Channel, declaring %q Exchange (%q)", exchangeType, exchange)
if err := channel.ExchangeDeclare(
exchange,     // name
exchangeType, // type
true,         // durable
false,        // auto-deleted
false,        // internal
false,        // noWait
nil,          // arguments
); err != nil {
return fmt.Errorf("Exchange Declare: %s", err)
}

// Reliable publisher confirms require confirm.select support from the
// connection.
if reliable {
log.Printf("enabling publishing confirms.")
if err := channel.Confirm(false); err != nil {
return fmt.Errorf("Channel could not be put into confirm mode: %s", err)
}

confirms := channel.NotifyPublish(make(chan amqp.Confirmation, 1))

defer confirmOne(confirms)
}

log.Printf("declared Exchange, publishing %dB body (%q)", len(body), body)
if err = channel.Publish(
exchange,   // publish to an exchange
routingKey, // routing to 0 or more queues
false,      // mandatory
false,      // immediate
amqp.Publishing{
Headers:         amqp.Table{},
ContentType:     "text/plain",
ContentEncoding: "",
Body:            []byte(body),
DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
Priority:        0,              // 0-9
},
); err != nil {
return fmt.Errorf("Exchange Publish: %s", err)
}

return nil
}

func confirmOne(confirms <-chan amqp.Confirmation) {
log.Printf("waiting for confirmation of one publishing")

if confirmed := <-confirms; confirmed.Ack {
log.Printf("confirmed delivery with delivery tag: %d", confirmed.DeliveryTag)
} else {
log.Printf("failed delivery of delivery tag: %d", confirmed.DeliveryTag)
}
}

ssl消费消息

package main

import (
"crypto/tls"
"crypto/x509"
"flag"
"fmt"
amqp "github.com/rabbitmq/amqp091-go"
"io/ioutil"
"log"
"time"
)

var (
uri          = flag.String("uri", "amqps://USERNAME:PASSWORD@10.10.33.196:5671", "AMQP URI")
exchange     = flag.String("exchange", "test-exchange", "Durable, non-auto-deleted AMQP exchange name")
exchangeType = flag.String("exchange-type", "direct", "Exchange type - direct|fanout|topic|x-custom")
queue        = flag.String("queue", "test-queue", "Ephemeral AMQP queue name")
bindingKey   = flag.String("key", "test-key", "AMQP binding key")
consumerTag  = flag.String("consumer-tag", "simple-consumer", "AMQP consumer tag (should not be blank)")
lifetime     = flag.Duration("lifetime", 5*time.Second, "lifetime of process before shutdown (0s=infinite)")
)

func init() {
flag.Parse()
}

func main() {
c, err := NewConsumer(*uri, *exchange, *exchangeType, *queue, *bindingKey, *consumerTag)
if err != nil {
log.Fatalf("%s", err)
}

if *lifetime > 0 {
log.Printf("running for %s", *lifetime)
time.Sleep(*lifetime)
} else {
log.Printf("running forever")
select {}
}

log.Printf("shutting down")

if err := c.Shutdown(); err != nil {
log.Fatalf("error during shutdown: %s", err)
}
}

type Consumer struct {
conn    *amqp.Connection
channel *amqp.Channel
tag     string
done    chan error
}

func NewConsumer(amqpsURI, exchange, exchangeType, queueName, key, ctag string) (*Consumer, error) {
c := &Consumer{
conn:    nil,
channel: nil,
tag:     ctag,
done:    make(chan error),
}

var err error

log.Printf("dialing %q", amqpsURI)

caCert, err := ioutil.ReadFile("D:\\codeTest\\amqp\\_examples\\ssl\\ca_certificate.pem")
if err != nil {
return nil, err
}

cert, err := tls.LoadX509KeyPair("D:\\codeTest\\amqp\\_examples\\ssl\\client_openstack_certificate.pem", "D:\\codeTest\\amqp\\_examples\\ssl\\client_openstack_key.pem")
if err != nil {
return nil, err
}

rootCAs := x509.NewCertPool()
rootCAs.AppendCertsFromPEM(caCert)

tlsConf := &tls.Config{
RootCAs:      rootCAs,
Certificates: []tls.Certificate{cert},
//ServerName:   "localhost", // Optional
InsecureSkipVerify: true,
}

c.conn, err = amqp.DialTLS(amqpsURI, tlsConf)
if err != nil {
return nil, fmt.Errorf("Dial: %s", err)
}

go func() {
fmt.Printf("closing: %s", <-c.conn.NotifyClose(make(chan *amqp.Error)))
}()

log.Printf("got Connection, getting Channel")
c.channel, err = c.conn.Channel()
if err != nil {
return nil, fmt.Errorf("Channel: %s", err)
}

log.Printf("got Channel, declaring Exchange (%q)", exchange)
if err = c.channel.ExchangeDeclare(
exchange,     // name of the exchange
exchangeType, // type
true,         // durable
false,        // delete when complete
false,        // internal
false,        // noWait
nil,          // arguments
); err != nil {
return nil, fmt.Errorf("Exchange Declare: %s", err)
}

log.Printf("declared Exchange, declaring Queue %q", queueName)
queue, err := c.channel.QueueDeclare(
queueName, // name of the queue
true,      // durable
false,     // delete when unused
false,     // exclusive
false,     // noWait
nil,       // arguments
)
if err != nil {
return nil, fmt.Errorf("Queue Declare: %s", err)
}

log.Printf("declared Queue (%q %d messages, %d consumers), binding to Exchange (key %q)",
queue.Name, queue.Messages, queue.Consumers, key)

if err = c.channel.QueueBind(
queue.Name, // name of the queue
key,        // bindingKey
exchange,   // sourceExchange
false,      // noWait
nil,        // arguments
); err != nil {
return nil, fmt.Errorf("Queue Bind: %s", err)
}

log.Printf("Queue bound to Exchange, starting Consume (consumer tag %q)", c.tag)
deliveries, err := c.channel.Consume(
queue.Name, // name
c.tag,      // consumerTag,
false,      // noAck
false,      // exclusive
false,      // noLocal
false,      // noWait
nil,        // arguments
)
if err != nil {
return nil, fmt.Errorf("Queue Consume: %s", err)
}

go handle(deliveries, c.done)

return c, nil
}

func (c *Consumer) Shutdown() error {
// will close() the deliveries channel
if err := c.channel.Cancel(c.tag, true); err != nil {
return fmt.Errorf("Consumer cancel failed: %s", err)
}

if err := c.conn.Close(); err != nil {
return fmt.Errorf("AMQP connection close error: %s", err)
}

defer log.Printf("AMQP shutdown OK")

// wait for handle() to exit
return <-c.done
}

func handle(deliveries <-chan amqp.Delivery, done chan error) {
for d := range deliveries {
log.Printf(
"got %dB delivery: [%v] %q",
len(d.Body),
d.DeliveryTag,
d.Body,
)
d.Ack(false)
}
log.Printf("handle: deliveries channel closed")
done <- nil
}