亲宝软件园·资讯

展开

RabbitMQ 入门 (Go) - 3. 模拟传感器,生成数据并发布

yangxu-pro 人气:0

现在,我们需要模拟传感器,生成数据,并发布到 RabbitMQ。

建立传感器项目

在 GOPATH src 下建立文件夹 sensors,使用 go mod init 初始化,并创建 main.go。

同时别忘了安装 amqp 的包:go get -u github.com/streadway/amqp

 

我们要生成一些模拟数据,生成数据有一定的范围(位于一个最大值和最小值之间),如下图:

 

因此,我们需要这样几个配置参数:

  1. 传感器的名称

  2. 传感器数据的更新频率

  3. 模拟生成数据的最大值

  4. 模拟生成数据的最小值

  5. 与前一次生成数据的差值的最大值(变化幅度的最大值)

 

设置命令行参数并读取

在这个项目里,我们需要通过命令行参数来传递配置,并在 Go 程序里面进行解析和读取。我们可以使用 os.Args 来搞这些命令行参数,但是更好的办法是使用 flag 这个包(其内部实现使用的也是 os.Args)。

 

我们先看代码:

  1. 第 5-9 行,我们声明了 5 个命令行参数。都是使用 flag 包下相应的函数实现的。

    1. 这几个命令行参数分别表示传感器名称、模拟数据的更新频率、模拟数据的最大值、最小值以及变化幅度的最大值。

    2. 这些命令行参数的类型分别是 string,uint,float64,float64,float64。

    3. 这些函数的参数都类似:

      1. 第一个参数是命令行参数的名称

      2. 第二个参数是命令行参数的默认值

      3. 第三个参数是参数的描述/帮助

  2. 在 main 函数里,我们调用     flat.Parse() 函数,就可以将命令行的参数值解析到 5-9 行声明变量里面。

 

我们测试一下,命令行输入 go run . --help,其结果如下:

 

生成模拟数据

要生成模拟传感器的数据,需要使用到 math/rand 和 time 这两个包。

先看代码:

  1. 第 17 行,我们需要一个 *rand.Rand 类型来生成随机数,它又需要一个源,这里使用 time.Now().UnixNano() 生成源,这样做的好处是因为这个时间纳秒数永远不会重复。

  2. 第 19 行,声明 value,它表示传感器的数值,在这先生成一个初始值。

  3. 第 20 行,是额定值,在这里也就是最大值最小值的中间平均值。

  4. 第 25 行,把更新频率(每秒更新的次数)转化为了两次更新之间的时间间隔(毫秒),并解析成 time.Duration 这个类型。

  5. 第 26 行,time.Tick 函数会返回一个 time 的 Channel,该函数会按照提供的时间间隔不断触发,并向这个 Channel 发送当前时间。

  6. 第 28 行,使用 for range 来处理 signal 这个 Channel,每次 Channel 中有数据传递过来,我们就使用 calcValue 这个函数来生成新的模拟数据。

  7. 第 29 行,把生成的最新数据打印一下即可。

 

calcValue 函数

生成模拟数据的逻辑是如果数据偏离额定值,那么尽量让下次生成的值向额定值靠拢。

这部分可根据自己的特定需求来实现,不必和我的相同。

先看代码:

  1. 第 35 行,声明了 maxStep 和 minStep 两个变量,表示本次更新相比上次所能够发生的最大变化和最小变化幅度。

  2. 第 36 - 42     行,区分当前值大于额定值或小于额定值两种情况,按不同的逻辑得出 maxStep 和 minStep

  3. 第 44 行,使用 maxStep     和 minStep 以及随机数生成新的 value 数据。

 

运行 sensors 项目

使用 go run . 运行,命令行参数使用默认值即可:

一切正常的话,它就会每秒钟生成 5 次数据。

 

如何运行多个传感器

生产环境中,通常会接收来自多个传感器的数据。

这里,我们让每个传感器都设置自己的路由 Key,所以 RabbitMQ 将会为每个 Key 创建一个 Queue:

但是这也会引起问题,就是之前章节里面的那个协调程序如何发现这些传感器呢?

首先,我们可以让每个传感器使用路由 Key 向一个所有传感器和协调程序都知晓的路径中发送一个消息。但这只能解决问题的一半,另一半我们以后再说。

 

将传感器数据发布到 RabbitMQ

创建传感器的消息类型

这里会使用到 encoding/gob 包。

看代码:

  • 在 sensors 包中创建 model 包,并建立 models.go 文件。

  • 在 models.go 的第 12 行,建立 SensorMessage 作为传感器传递消息的类型,里面包含三个字段分别是传感器名称、数值和时间戳。

  • 很显然我们不能把 Go 的 struct 类型直接扔到 RabbitMQ 里面,但我们项目中的各种客户端只涉及到 Go 语言,所以在这里我使用 Go 语言的 gob 来对消息进行编码,这样会更高效一些。如果这个项目是跨语言的我可能会使用 JSON 或 Protocol Buffers。

  • 在 model 包的 init 函数里面,需要使用 gob 包的 Register 函数把将要编码的类型进行注册,这样依赖于这个包的其它 Go 程序就可以把     SensorMessage 这个类型的消息对象发送过去了

 

建立 Queue 相关的工具包

建立 tools 包,并建立 queuetools.go 文件,其内容如下:

代码内容与之前的项目类似,就不解释了。

 

发布传感器数据到 RabbitMQ

这里还会使用到 bytes 包。

回到 main.go,修改代码:

  1. 前面添加了获取 Channel 和 Queue 的代码。其中第 37 行比较重要,因为我们不能保证在程序运行时,使用 Queue 名称作为路由 Key 的 Queue 存在,而使用 GetQueue 函数,就可以保证这个 Queue 会被正确的设置,并准备好被我们使用了。

  2. 第 42 行,使用 bytes 包创建了一个 *bytes.Buffer,它用来来承载编码后的数据,这个 Buffer 可以重复利用,所以实在 for range 的外部声明的。

    1. 但是每次使用 Buffer 都需要进行重置,也就是第 53 行的作用,这样以前的数据就会被移除,Buffer 的指针会回到初始位置。

  3. 第 43 行,使用 gob 和 Buffer 来创建编码器 。

  4. 第 54 行,使用 编码器的 Encode 方法对消息进行编码。

  5. 第 56 行,创建要发送给 RabbitMQ 的消息(amqp.Publishing 类型),这里只需要填写 Body 字段即可,其它的字段根据自己的需求选填即可。

  6. 第 60 行,使用 Channel 来发布消息,这里使用的是默认的 Exchange,路由 key 就是 Queue 的名字,最后一个参数就是发布的消息。

 

运行程序

运行 sensors 包:

 

打开控制台:

可以看到发送频率确实是每秒 5 次。

 

打开 sensor Queue:

目前已经有 384 条消息了,都没有被发送。

 

随便点开一个消息查看其内容:

可以看到 Body 应该是 Base64 编码的。因为 gob 编码器使用的是二进制消息格式,尽可能的高效,所以在控制台里面它没有一个有意义的表述展示。

 

然后,先停止运行程序。

 

传感器上线时通知协调程序

最后我们就来处理上面那个问题:当传感器上线的时候,得让协调程序知道,并发送数据。

因为每个传感器都创建了一个自己的 Queue,所以在没有帮助的情况下,协调程序将无法有效知道这些传感器。

这个问题实际上具体需要做两件事,我们先来做第一件事:

多个传感器他们 Queue 的名称是不一样的,是动态的,所以我们需要一个大家都知道的 Queue,它用来将每个新创建的传感器的 Queue 名称发送给协调程序。

 

首先,在 queuetools.go 里面添加这个 Queue 的名称,使用一个常量保存:

 

然后,在 main.go 里,使用这个名称创建一个 Queue,并将传感器的 Queue 的名称发布上去:

 

再次运行 sensor 包

打开控制台:

可以看到 SensorList Queue 出现了。

 

进入到 SensorList Queue,看它的 Message:

 

可以看到当前这一个传感器的名字 sensor 就在里面。

 
 

加载全部内容

相关教程
猜你喜欢
用户评论