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

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

RabbitMQ搭建

docker

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

生产者
package main

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

func main() {
  var url = flag.String("url", "amqp://admin:admin@localhost:5672/my_vhost", "RabbitMQ server URL")
  var exchange = flag.String("exchange", "logs", "Exchange name")
  var key = flag.String("key", "log", "Routing key")

  flag.Parse()

  // 连接到RabbitMQ服务器
  conn, err := amqp.Dial(*url)
  if err != nil {
    log.Fatalf("Failed to connect to RabbitMQ: %v", err)
  }
  defer conn.Close()

  // 创建一个通道
  ch, err := conn.Channel()
  if err != nil {
    log.Fatalf("Failed to open a channel: %v", err)
  }
  defer ch.Close()

  // 声明一个交换机
  err = ch.ExchangeDeclare(
    *exchange, // name: 交换机名称
    "fanout",  // kind: 交换机类型
    true,      // durable: 是否持久化
    false,     // autoDelete: 没有队列绑定时是否自动删除
    false,     // internal: 是否是内部交换机
    false,     // noWait: 是否需要等待服务器响应
    nil,       // args: 其他参数
  )
  if err != nil {
    log.Fatalf("Failed to declare an exchange: %v", err)
  }

  // 发送消息
  body := "Hello World!" + fmt.Sprintf(time.Now().String())

  for i := 0; i < 20; i++ {
    body = strconv.Itoa(i) + body
    err = ch.Publish(
      *exchange, // 交换机名称
      *key,      // 路由键
      false,     // 强制发布
      false,     // 立即发布
      amqp.Publishing{
        ContentType:  "text/plain",
        DeliveryMode: amqp.Persistent,
        Body:         []byte(body),
        Expiration:   "10000", // 3000 3秒
      })
  }

  if err != nil {
    log.Fatalf("Failed to publish a message: %v", err)
  }

  fmt.Printf(" [x] Sent %s", body)
}

消费者
package main

import (
  "flag"
  "fmt"
  "log"

  amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
  var url = flag.String("url", "amqp://admin:admin@localhost:5672/my_vhost", "RabbitMQ server URL")
  var exchange = flag.String("exchange", "logs", "Exchange name")
  var key = flag.String("key", "log", "Routing key")

  flag.Parse()

  // 连接到RabbitMQ服务器
  conn, err := amqp.Dial(*url)
  if err != nil {
    log.Fatalf("Failed to connect to RabbitMQ: %v", err)
  }
  defer conn.Close()

  // 创建一个通道
  ch, err := conn.Channel()
  if err != nil {
    log.Fatalf("Failed to open a channel: %v", err)
  }
  defer ch.Close()

  // 声明一个交换机
  err = ch.ExchangeDeclare(
    *exchange, // name: 交换机名称
    "fanout",  // kind: 交换机类型
    true,      // durable: 是否持久化
    false,     // autoDelete: 没有队列绑定时是否自动删除
    false,     // internal: 是否是内部交换机
    false,     // noWait: 是否需要等待服务器响应
    nil,       // args: 其他参数
  )
  if err != nil {
    log.Fatalf("Failed to declare an exchange: %v", err)
  }

  // 声明一个队列
  q, err := ch.QueueDeclare(
    "queue01", // 随机生成队列名称
    true,      // 持久化
    false,     // 删除
    false,     // 独占
    false,     // 不等消息
    nil,       // 其他参数
  )
  if err != nil {
    log.Fatalf("Failed to declare a queue: %v", err)
  }

  // 绑定队列到交换机
  err = ch.QueueBind(
    q.Name,    // 队列名称
    *key,      // 路由键
    *exchange, // 交换机名称
    false,     // 现在绑定
    nil,       // 其他参数
  )
  if err != nil {
    log.Fatalf("Failed to bind a queue: %v", err)
  }

  // 接收消息
  msgs, err := ch.Consume(
    q.Name,       // 队列名称
    "consumer01", // 消费者标签
    false,        // 自动ack
    false,        // 不独占
    false,        // 不等消息
    false,        // 不从服务器获取消息
    nil,          // 其他参数
  )
  if err != nil {
    log.Fatalf("Failed to register a consumer: %v", err)
  }

  fmt.Printf(" [*] Waiting for messages. To exit press CTRL+C\n")
  for d := range msgs {
    // 输出接收到的消息
    fmt.Printf(" [x] Received %s\n", d.Body)
    err = ch.Ack(d.DeliveryTag, true)
    if err != nil {
      log.Fatalf("Failed to ack message: %v", err)
    }
  }
}

可视化

看板

http://localhost:15672/

账户密码

admin
admin

消费进度

http://localhost:15672/#/queues

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
2月前
|
消息中间件 存储 数据库
RocketMQ 流存储解析:面向流场景的关键特性与典型案例
RocketMQ 流存储解析:面向流场景的关键特性与典型案例
88461 0
|
2月前
|
Java API 网络架构
关于 Spring Integration 你知道多少,包含集成MQTT案例讲述及源码3
关于 Spring Integration 你知道多少,包含集成MQTT案例讲述及源码
914 0
关于 Spring Integration 你知道多少,包含集成MQTT案例讲述及源码3
|
2月前
|
消息中间件 Java API
RocketMQ事务消息, 图文、源码学习探究~
介绍 RocketMQ是阿里巴巴开源的分布式消息中间件,它是一个高性能、低延迟、可靠的消息队列系统,用于在分布式系统中进行异步通信。 从4.3.0版本开始正式支持分布式事务消息~ RocketMq事务消息支持最终一致性:在普通消息基础上,支持二阶段的提交能力。将二阶段提交和本地事务绑定,实现全局提交结果的一致性。 原理、流程 本质上RocketMq的事务能力是基于二阶段提交来实现的 在消息发送上,将二阶段提交与本地事务绑定 本地事务执行成功,则事务消息成功,可以交由Consumer消费 本地事务执行失败,则事务消息失败,Consumer无法消费 但是,RocketMq只能保证本地事务
|
2月前
|
消息中间件 Apache 开发工具
RocketMQ-初体验RocketMQ(08)-IDEA拉取调测RocketMQ源码
RocketMQ-初体验RocketMQ(08)-IDEA拉取调测RocketMQ源码
46 0
|
17天前
|
数据采集 监控 物联网
MQTT协议在智能制造中的应用案例与效益分析
【6月更文挑战第8天】MQTT协议在智能制造中的应用案例与效益分析
61 1
|
2月前
|
消息中间件 小程序 网络性能优化
蓝易云 - 直播小程序源码有用的协议知识:MQTT协
在直播小程序源码中,MQTT协议可以用于实现实时消息推送,如弹幕、聊天消息、礼物信息等。通过使用MQTT协议,可以确保消息的实时性和可靠性,从而提高用户体验。
61 0
|
2月前
|
消息中间件 Java 调度
【深度挖掘RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行调度的流程(Pull模式)
【深度挖掘RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行调度的流程(Pull模式)
31 1
|
2月前
|
消息中间件 Java RocketMQ
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」抽丝剥茧贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-下)
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」抽丝剥茧贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-下)
24 1
|
2月前
|
消息中间件 存储 NoSQL
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)
37 1
|
2月前
|
消息中间件 存储 Kafka
【深入浅出 RocketMQ原理及实战】「底层源码挖掘系列」透彻剖析贯穿一下RocketMQ和Kafka索引设计原理和方案
【深入浅出 RocketMQ原理及实战】「底层源码挖掘系列」透彻剖析贯穿一下RocketMQ和Kafka索引设计原理和方案
76 1