关于消息队列:消息队列RabbitMQ简单使用

装置应用

• 地址：http://www.rabbitmq.com/
• 须要依据不同的版本抉择不同的erlang

装置erlang

装置RabbitMQ首先须要装置erlang环境，依据GitHub上erlang中README配置yum源

将下面的内容写入红线标注的文件中保留退出，而后装置erlang

yum update -y
yum install -y erlang-23.3.4

装置RabbitMQ

下载实现RabbitMQ的安装包后执行上面的命令装置

yum install rabbitmq-server-3.8.19-1.el7.noarch.rpm

装置实现后能够看到有6个相干的可执行程序

其中外围是rabbitmq-server

RabbitMQ的应用

启动

rabbitmq-server start & # 后盾启动

启动RabbitMQ图形化治理界面

RabbitMQ的图形化治理界面模式是没有关上的，须要咱们手动去关上，rabbitmq-plugins就是治理各种插件的工具

rabbitmq-plugins list # 查看所有插件
rabbitmq-plugins enable rabbitmq_management # 关上图形化界面
rabbitmq-plugins disable rabbitmq_management # 敞开图形化界面

拜访图形化治理界面

关上图形化治理界面插件后就能够通过本机浏览器拜访RabbitMQ了：http://127.0.0.1:15672/ 用户名和明码都是guest

如果要在其余机器上拜访该界面则须要配置用户名和明码，权限等

rabbitmqctl add_user Guest Guest # 增加用户
rabbitmqctl set_user_tags Guest administrator # 赋予该用户管理员角色
rabbitmqctl set_permissions -p / Guest ".*" ".*" ".*" # 给该用户赋予权限

这样就能够在其余机器上拜访RabbitMQ的治理界面了

上面通过Go语言实现一个简略的音讯队列的利用

consumer.go

package main

import (
   "fmt"
   "github.com/streadway/amqp"
)

//消费者

func main() {
   //1.建设连贯
   connection, err := amqp.Dial("amqp://Guest:Guest@192.168.27.139:5672")
   if err != nil {
      panic(err)
   }
   defer connection.Close()
   //2.设置通道
   ch, err := connection.Channel()
   if err != nil {
      panic(err)
   }
   //3.定义exchange
   //err = ch.ExchangeDeclare("exchange-name", "direct", false, false, false, false, nil)
   err = ch.ExchangeDeclare("exchange-name-topic", "topic", false, false, false, false, nil)
   if err != nil {
      panic(err)
   }
   //4.定义队列 queue
   queue, err := ch.QueueDeclare("", false, false,false,false, nil)
   if err != nil {
      panic(err)
   }
   //5.绑定 bind
   //err = ch.QueueBind(queue.Name, "direct_key", "exchange-name", false, nil)
   //topic 含糊匹配
   err = ch.QueueBind(queue.Name, "topic.#", "exchange-name-topic", false, nil)
   if err != nil {
      panic(err)
   }
   //6.接管音讯
   consume_msg, err := ch.Consume(queue.Name, "", false, false, false, false, nil)
   if err != nil {
      panic(err)
   }
   //7.打印消息
   //msg := <-consume_msg
   //fmt.Printf("received msg: %s\n", msg.Body)
   for msg := range consume_msg {
      fmt.Printf("received msg: %s\n", msg.Body)
   }
}

producer.go

package main

import "github.com/streadway/amqp"

//生产者

func main() {
   //1.建设连贯
   connection, err := amqp.Dial("amqp://Guest:Guest@192.168.27.139:5672")
   if err != nil {
      panic(err)
   }
   defer connection.Close()
   //2.设置通道
   ch, err := connection.Channel()
   if err != nil {
      panic(err)
   }
   //3.定义exchange
   //err = ch.ExchangeDeclare("exchange-name", "direct", false, false, false, false, nil)
   err = ch.ExchangeDeclare("exchange-name-topic", "topic", false, false, false, false, nil)
   if err != nil {
      panic(err)
   }
   //4.定义队列 queue
   //queue, err := ch.QueueDeclare("", false, false,false,false, nil)
   //if err != nil {
   // panic(err)
   //}
   //5.绑定 bind
   //6.定义公布的音讯
   //msg := "hello"
   //7.公布音讯
   //err = ch.Publish("exchange-name", "direct_key", false, false, amqp.Publishing{Body: []byte(msg)})
   err = ch.Publish("exchange-name-topic", "topic.first", false, false, amqp.Publishing{Body: []byte("topic1")})
   err = ch.Publish("exchange-name-topic", "TOPIC.eg", false, false, amqp.Publishing{Body: []byte("topic-eg")})
   err = ch.Publish("exchange-name-topic", "topic.last", false, false, amqp.Publishing{Body: []byte("topic2")})
   if err != nil {
      panic(err)
   }
}