rabbitmq消息队列——"Hello World!"

RabbitMQ

一、”Hello World!”

1、简介:

         RabbitMQ是一种消息中间件,主要思想很简单:接收消息并转发。你可以将它设想为一个邮局:你往里面发送邮件并确保邮件能实际运达,RabbitMQ好比这里的邮箱、邮局和邮递员的角色。

         RabbitMQ和邮局的一个主要区别是,RabbitMQ仅仅接收、存储、转发这些数据包裹——message。

先来看下RabbitMQ中的一些关键术语:

a)、生产(者):除了发送什么意义都没有。一个发送消息的应用就是一个生产者,使用如下描述:

b)、队列:储存消息的“容器”,可以储存任意多的message——本质上是一个无限长度的缓冲区,多个生产者可以将消息发送至同一队列,多个消费者也可以从同一队列中接收消息。队列使用如下描述,”queue_name”是该队列的名称:

c)、消费(者):一个消费者就好比一个用来等待接收消息的程序。使用如下来描述:

2、”Hello World!”(使用Go RabbitMQ客户端)

         这节我们将使用Go写两个小程序:一个生产者用来发送单一消息,一个消费者用来接收这些消息并打印。图示如下:

备注:这里使用“amqp”包,自行安装:
go get github.com/streadway/amqp

发送:

         首先创建一个send.go和receiver.go分别用来发送和接收,发送方会链接RabbitMQ服务器、发送消息然后退出。

在send.go,首先导入相关包:

package main

import (
  "fmt"
  "log"

  "github.com/streadway/amqp"
)

再加一个出错处理类:

func failOnError(err error, msg string) {
  if err != nil {
    log.Fatalf("%s: %s", msg, err)
    panic(fmt.Sprintf("%s: %s", msg, err))
  }
}

连接到服务器:

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")  //一般默认端口5672
failOnError(err, "Failed to connect to RabbitMQ")
defer conn.Close()

这里的conn对象是抽象了底层的socket链接,在conn的基础上我们可以创建多个channel(通道,一个conn可以创建多个channel,使用channel节省了tcp资源,后续的很多操作如:队列声明、消息声明发送、交换器声明等都是在channel基础上操作的),接下来是产生一个channel:

ch, err := conn.Channel()
failOnError(err, "Failed to open a channel")
defer ch.Close()

为了发送消息到队列,我们还应该声明一个队列,并将消息发送给它:

q, err := ch.QueueDeclare(
  "hello", // name 队列名称
  false,   // durable    是否持久化,默认为false
  false,   // delete when unused 队列无订阅时是否自动删除队列
  false,   // exclusive 是否队列私有,私有后仅有一个应用可以连接
  false,   // no-wait
  nil,     // arguments
)
failOnError(err, "Failed to declare a queue")

body := "hello"    //消息主体
err = ch.Publish(         //发送消息
  "",     // exchange    //指定消息发送的交换器名称
  q.Name, // routing key   //路由键
  false,  // mandatory
  false,  // immediate
  amqp.Publishing {
    ContentType: "text/plain",    //消息类型:文本消息
    Body:        []byte(body),  //消息体,理论上可以发送任意类型数据,因为是byte类型
  })
failOnError(err, "Failed to publish a message")

至此发送方代码完毕!

接收:

接收跟发送不同的是,接收端一直运行监听发送端消息发送并打印输出,接收端的模型如下:

在这里,我们仍然使用send.go中的逻辑执行,首先是链接服务器,其次是声明channel和队列(可以防止接收端启动时发送端还没有启动的情况),主要代码如下:

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
failOnError(err, "Failed to connect to RabbitMQ")
defer conn.Close()

ch, err := conn.Channel()
failOnError(err, "Failed to open a channel")
defer ch.Close()

q, err := ch.QueueDeclare(
  "hello", // name
  false,   // durable
  false,   // delete when usused
  false,   // exclusive
  false,   // no-wait
  nil,     // arguments
)
failOnError(err, "Failed to declare a queue")

这里声明的队列名称就是send.go中声明的队列,然后从该队列中读取消息并打印:

msgs, err := ch.Consume(
  q.Name, // queue
  "",     // consumer
  true,   // auto-ack
  false,  // exclusive
  false,  // no-local
  false,  // no-wait
  nil,    // args
)
failOnError(err, "Failed to register a consumer")

forever := make(chan bool)

go func() {
  for d := range msgs {
    log.Printf("Received a message: %s", d.Body)
  }
}()
log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever

至此,所有代码书写完毕,完整版代码如下:

发送:

package main

import (
    "fmt"
    "log"

    "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s:%s", msg, err)
        panic(fmt.Sprintf("%s:%s", msg, err))
    }
}

func main() {
    //链接rabbitMQ服务器
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "fail to connect to rabbitmq server")
    defer conn.Close()

    //声明一个channel
    ch, err := conn.Channel()
    failOnError(err, "fail to open a channel")
    defer ch.Close()

    //根据channel声明一个队列
    q, err := ch.QueueDeclare("queue_name", false, false, false, false, nil)
    failOnError(err, "fail to define a queue")

    //使用channel直接发送消息至队列
    body := "hello queue!"
    err = ch.Publish("", q.Name, false, false, amqp.Publishing{
        ContentType: "text/plain",
        Body:        []byte(body),
    })
    failOnError(err, "fail to publish the message")
}

接收:

// rabbitmq_1.receiver project main.go
package main

import (
    "fmt"
    "log"

    "github.com/streadway/amqp"
)

func main() {
    //链接RabbitMQ服务器
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    //声明一个channel
    ch, err := conn.Channel()
    failOnError(err, "failed to open a channel")
    defer ch.Close()

    //使用channel声明一个队列
    q, err := ch.QueueDeclare("queue_name", false, false, false, false, nil)
    failOnError(err, "failed to declare a queue")

    //注册一个消费者
    msgs, err := ch.Consume(q.Name, "", true, false, false, false, nil)
    failOnError(err, "failed to register a consumer")

    //定义一个never型chan,用于防止进程退出
    var forever chan bool = make(chan bool, 0)
    //开启一个channel,实时打印channel中的消息
    go func() {
        for d := range msgs {
            log.Printf("received a message:%s", d.Body)
        }
    }()

    log.Printf("press Ctrl+c to exit!")
    <-forever
}

func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
        panic(fmt.Sprintf("%s: %s", msg, err))
    }
}

然后我们使用RabbitMQ自带的管理工具查看mq运行情况:

首先要开启管理工具:
rabbitmq-plugins enable rabbitmq_management

然后浏览器访问地址:http://localhost:15672/,这里输入默认用户名密码:guest/guest,进去后界面如下:

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Kirito的技术分享

深入理解 RPC 之集群篇

上一篇文章分析了服务的注册与发现,这一篇文章着重分析下 RPC 框架都会用到的集群的相关知识。 集群(Cluster)本身并不具备太多知识点,在分布式系统中,...

3139
来自专栏哲学驱动设计

asp.net MVC 应用程序的生命周期

  首先我们知道http是一种无状态的请求,他的生命周期就是从客户端浏览器发出请求开始,到得到响应结束。那么MVC应用程序从发出请求到获得响应,都做了些什么呢?

481
来自专栏进击的程序猿

orm 系列 之 Eloquent演化历程1

Eloquent是laravel中的orm,采取的是active record的设计模式,里面的对象不仅包括领域逻辑,还包括了数据库操作,但是大家平时使用的时候...

623
来自专栏张善友的专栏

更强悍的Silverlight: WCF RIA Services

如果你安装了 VS 2010 的 Silverlight 4 开发工具,会发现一项重量级的安装项目,WCF RIA Services,用于Silverlight...

18010
来自专栏移动端开发

iOS 从实际出发理解多线程

前言 ----       多线程很多开发者多多少少相信也都有了解,以前有些东西理解的不是很透,慢慢的积累之后,这方面的东西也需要自己好好的总结一下。多线程从我...

1957
来自专栏Adamshuang 技术文章

Guava Cache -- Java 应用缓存神器

Guava 作为Google开源Java 库中的精品成员,在性能、功能上都十分出色,本文将从实际使用的角度,来对Guava进行讲解。

5747
来自专栏哲学驱动设计

asp.net MVC 应用程序的生命周期

  首先我们知道http是一种无状态的请求,他的生命周期就是从客户端浏览器发出请求开始,到得到响应结束。那么MVC应用程序从发出请求到获得响应,都做了些什么呢?...

1629
来自专栏java一日一条

面试中问的话题Spring工作原理

Spring工作原理 内部最核心的就是IOC了, 动态注入,让一个对象的创建不用new了,可以自动的生产,这其实就是利用java里的反射 反射其实就是在...

634
来自专栏闻道于事

quartz定时任务,已过期的内容自动下线

概念: Quartz是一个开源的作业调度框架,可以让计划的程序任务一个预定义的日期和时间运行。Quartz可以用来创建简单或复杂的日程安排执行几十,几百,甚至是...

2645
来自专栏企鹅号快讯

WebApi client 的面向切面编程

.Net的面向切面编程 .Net的服务端应用AOP很常见,在Asp.net MVC与Asp.net WebApi等新框架里到处都有AOP的影子,我们可以把一个服...

18111

扫码关注云+社区