Rabbitmq 搭建使用案例 [附源码]

Rabbitmq 搭建使用案例

@
目录

• RabbitMQ搭建
  
  
  
  • docker
    
  
  
• 代码
  
  
  
  • golang
    
    
    
    • 生产者
      
    • 消耗者
      
    
    
  
  
• 可视化
  
  
  
  • 消耗进度
    
  
  

RabbitMQ搭建

docker

1. docker run -d --hostname rabbitmq --name rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -e RABBITMQ_DEFAULT_VHOST=my_vhost -e RABBITMQ_MANAGEMENT_PLUGIN=rabbitmq_management_agent -p 15672:15672 -p 5672:5672 rabbitmq:management

复制代码

代码

golang

生产者

1. package main
3. import (
4.         "flag"
5.         "fmt"
6.         amqp "github.com/rabbitmq/amqp091-go"
7.         "log"
8.         "strconv"
9.         "time"
10. )
12. func main() {
13.         var url = flag.String("url", "amqp://admin:admin@localhost:5672/my_vhost", "RabbitMQ server URL")
14.         var exchange = flag.String("exchange", "logs", "Exchange name")
15.         var key = flag.String("key", "log", "Routing key")
17.         flag.Parse()
19.         // 连接到RabbitMQ服务器
20.         conn, err := amqp.Dial(*url)
21.         if err != nil {
22.                 log.Fatalf("Failed to connect to RabbitMQ: %v", err)
23.         }
24.         defer conn.Close()
26.         // 创建一个通道
27.         ch, err := conn.Channel()
28.         if err != nil {
29.                 log.Fatalf("Failed to open a channel: %v", err)
30.         }
31.         defer ch.Close()
33.         // 声明一个交换机
34.         err = ch.ExchangeDeclare(
35.                 *exchange, // name: 交换机名称
36.                 "fanout",  // kind: 交换机类型
37.                 true,      // durable: 是否持久化
38.                 false,     // autoDelete: 没有队列绑定时是否自动删除
39.                 false,     // internal: 是否是内部交换机
40.                 false,     // noWait: 是否需要等待服务器响应
41.                 nil,       // args: 其他参数
42.         )
43.         if err != nil {
44.                 log.Fatalf("Failed to declare an exchange: %v", err)
45.         }
47.         // 发送消息
48.         body := "Hello World!" + fmt.Sprintf(time.Now().String())
50.         for i := 0; i < 20; i++ {
51.                 body = strconv.Itoa(i) + body
52.                 err = ch.Publish(
53.                         *exchange, // 交换机名称
54.                         *key,      // 路由键
55.                         false,     // 强制发布
56.                         false,     // 立即发布
57.                         amqp.Publishing{
58.                                 ContentType:  "text/plain",
59.                                 DeliveryMode: amqp.Persistent,
60.                                 Body:         []byte(body),
61.                                 Expiration:   "10000", // 3000 3秒
62.                         })
63.         }
65.         if err != nil {
66.                 log.Fatalf("Failed to publish a message: %v", err)
67.         }
69.         fmt.Printf(" [x] Sent %s", body)
70. }

复制代码

消耗者

1. package main
3. import (
4.         "flag"
5.         "fmt"
6.         "log"
8.         amqp "github.com/rabbitmq/amqp091-go"
9. )
11. func main() {
12.         var url = flag.String("url", "amqp://admin:admin@localhost:5672/my_vhost", "RabbitMQ server URL")
13.         var exchange = flag.String("exchange", "logs", "Exchange name")
14.         var key = flag.String("key", "log", "Routing key")
16.         flag.Parse()
18.         // 连接到RabbitMQ服务器
19.         conn, err := amqp.Dial(*url)
20.         if err != nil {
21.                 log.Fatalf("Failed to connect to RabbitMQ: %v", err)
22.         }
23.         defer conn.Close()
25.         // 创建一个通道
26.         ch, err := conn.Channel()
27.         if err != nil {
28.                 log.Fatalf("Failed to open a channel: %v", err)
29.         }
30.         defer ch.Close()
32.         // 声明一个交换机
33.         err = ch.ExchangeDeclare(
34.                 *exchange, // name: 交换机名称
35.                 "fanout",  // kind: 交换机类型
36.                 true,      // durable: 是否持久化
37.                 false,     // autoDelete: 没有队列绑定时是否自动删除
38.                 false,     // internal: 是否是内部交换机
39.                 false,     // noWait: 是否需要等待服务器响应
40.                 nil,       // args: 其他参数
41.         )
42.         if err != nil {
43.                 log.Fatalf("Failed to declare an exchange: %v", err)
44.         }
46.         // 声明一个队列
47.         q, err := ch.QueueDeclare(
48.                 "queue01", // 随机生成队列名称
49.                 true,      // 持久化
50.                 false,     // 删除
51.                 false,     // 独占
52.                 false,     // 不等消息
53.                 nil,       // 其他参数
54.         )
55.         if err != nil {
56.                 log.Fatalf("Failed to declare a queue: %v", err)
57.         }
59.         // 绑定队列到交换机
60.         err = ch.QueueBind(
61.                 q.Name,    // 队列名称
62.                 *key,      // 路由键
63.                 *exchange, // 交换机名称
64.                 false,     // 现在绑定
65.                 nil,       // 其他参数
66.         )
67.         if err != nil {
68.                 log.Fatalf("Failed to bind a queue: %v", err)
69.         }
71.         // 接收消息
72.         msgs, err := ch.Consume(
73.                 q.Name,       // 队列名称
74.                 "consumer01", // 消费者标签
75.                 false,        // 自动ack
76.                 false,        // 不独占
77.                 false,        // 不等消息
78.                 false,        // 不从服务器获取消息
79.                 nil,          // 其他参数
80.         )
81.         if err != nil {
82.                 log.Fatalf("Failed to register a consumer: %v", err)
83.         }
85.         fmt.Printf(" [*] Waiting for messages. To exit press CTRL+C\n")
86.         for d := range msgs {
87.                 // 输出接收到的消息
88.                 fmt.Printf(" [x] Received %s\n", d.Body)
89.                 err = ch.Ack(d.DeliveryTag, true)
90.                 if err != nil {
91.                         log.Fatalf("Failed to ack message: %v", err)
92.                 }
93.         }
94. }

复制代码

可视化

看板
http://localhost:15672/
账户密码

1. admin
2. admin

复制代码

消耗进度

http://localhost:15672/#/queues

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。