RabbitMQ play
Anonymous
golang
02/13/2024 6:41 PM
3.8 KB
15
Indexable
package main
import (
"fmt"
"github.com/streadway/amqp"
"log"
"time"
)
// RabbitMQ server connection URL
var (
amqpURI = "amqp://guest:guest@localhost:5672/"
queueName = "exampleQueue"
)
func main() {
//ch := make(chan int)
//close(ch)
//for {
// select {
// case <-ch:
// fmt.Println("hello, world!")
// time.Sleep(10 * time.Second)
// default:
// fmt.Println("good night!")
// time.Sleep(10 * time.Second)
// }
//}
//ch := make(chan int, 2)
//fmt.Println(<-ch)
//close(ch)
//ch2 := make(chan int, 2)
//fmt.Println(<-ch2)
// Establish a connection to RabbitMQ server
connection, err := amqp.Dial(amqpURI)
if err != nil {
log.Fatalf("Failed to connect to RabbitMQ: %v\n", err)
}
defer connection.Close()
closeConnChan := make(chan *amqp.Error)
connection.NotifyClose(closeConnChan)
go func() {
<-closeConnChan
connection.Close()
}()
fmt.Println("Connected to RabbitMQ!")
connection.Channel()
channel, err := connection.Channel()
if err != nil {
log.Fatalf("Failed to open a channel: %v", err)
}
defer channel.Close()
channelCloseCh := make(chan *amqp.Error)
channel.NotifyClose(channelCloseCh)
go func() {
<-channelCloseCh
fmt.Println("channel closed!")
}()
// Declare a queue
queue, err := channel.QueueDeclare(
queueName, // name
true, // durable
false, // delete when unused
false, // exclusive
false, // no-wait
nil, // arguments
)
if err != nil {
log.Fatalf("Failed to declare a queue: %v", err)
}
fmt.Println("queue.Name =", queue.Name)
// Consume messages from the queue
messages, err := channel.Consume(
queueName, // queue
"Abdrasul", // consumer
true, // auto-acknowledge
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
if err != nil {
log.Fatalf("Failed to consume messages: %v", err)
}
fmt.Println("Waiting for messages. To exit press CTRL+C")
// Block and wait for messages
for message := range messages {
fmt.Printf("Received a message: %s\n", message.Body)
}
go func() {
for {
select {
case err := <-closeConnChan:
fmt.Printf("rabbitmq connection error - %v\n", err)
connection.Close()
time.Sleep(time.Second * 10)
for {
var err2 error
connection, err2 = reopenRabbitMQConnection()
if err2 == nil {
fmt.Println("notify?")
closeConnChan = make(chan *amqp.Error)
connection.NotifyClose(closeConnChan)
break
}
time.Sleep(10 * time.Second)
}
}
}
}()
fmt.Println("Night night!")
// Further logic for interacting with RabbitMQ, e.g., creating channels, declaring queues, etc., can be added here.
}
func reopenRabbitMQConnection() (*amqp.Connection, error) {
connection, err := amqp.Dial(amqpURI)
if err != nil {
log.Printf("Failed to connect to RabbitMQ: %v\n", err)
return nil, err
}
return connection, nil
}
Editor is loading...
Leave a Comment