设为首页 收藏本站
查看: 1177|回复: 0

[经验分享] 关于openstack的Rabbitmq安装

[复制链接]

尚未签到

发表于 2018-6-1 07:18:55 | 显示全部楼层 |阅读模式
RABBITMQ服务

  一、RATTITMQ的概念
RabbitMQ 是实现了高级消息队列协议(AMQP)的开源消息代理软件(亦称面向消息的中间件)。
AMQP是一个定义了在应用或者组织之间传送消息的协议的开放标准 (an open standard for passing business messages between applications or organizations)。AMQP 目标在于解决在两个应用之间传送消息存在的下列问题:
· 网络是不可靠的 =>消息需要保存后再转发并有出错处理机制
· 与本地调用相比,网络速度慢 =>得异步调用
· 应用之间是不同的(比如不同语言实现、不同操作系统等) =>得与应用无关
· 应用会经常变化 =>同上
AMQP 使用异步的、应用对应用的、二进制数据通信来解决这些问题。
RabbitMQAMQP 的一种实现,它包括Server (服务器端)、Client (客户端) 和 Plugins (插件)。RabbitMQ 服务器是用 Erlang 语言编写的
1.1 RabbitMQ 的概念非常清晰、简洁
其基本概念参见下图
DSC0000.png
简单说明如下:
· Message (消息)RabbitMQ 转发的二进制对象,包括Headers(头)、Properties (属性)和 Data (数据),其中数据部分不是必要的。具体见 1.2 部分的描述。
· Producer(生产者): 消息的生产者,负责产生消息并把消息发到交换机 Exhange的应用。
· Consumer (消费者):使用队列 Queue Exchange 中获取消息的应用。
· Exchange (交换机):负责接收生产者的消息并把它转到到合适的队列 Queue 。下面有 1.3 部分描述。
· Queue (队列):一个存储Exchange 发来的消息的缓冲,并将消息主动发送给Consumer,或者 Consumer 主动来获取消息。见 1.4 部分的描述
· Binding (绑定):队列 和 交换机 之间的关系。Exchange 根据消息的属性和 Binding 的属性来转发消息。绑定的一个重要属性是 binding_key
· Connection (连接)和 Channel (通道):生产者和消费者需要和 RabbitMQ 建立 TCP 连接。一些应用需要多个connection,为了节省TCP 连接,可以使用 Channel,它可以被认为是一种轻型的共享 TCP 连接的连接。连接需要用户认证,并且支持 TLS (SSL)。连接需要显式关闭。
· Virtual Host (虚拟主机)RabbitMQ 用来进行资源隔离的机制。一个虚机主机会隔离用户、exchangequeue 等。默认的虚拟主机为 "/"
1.2 关于消息 message
消息结构:
DSC0001.png
消息的几个重要属性:
· routing_key:Direct Topic 类型的 exchange 会根据本属性来转发消息
· delivery_mode: 将其值设置为 2 将用于消息的持久化,持久化的消息会被保存到磁盘上来防止其丢失。下面章节 3 有描述。
· reply_to:一般用来表示RPC实现中客户端的回调队列的名字。下面章节 4 有描述。
· correlation_id:用于使用 RabbitMQ 来实现 RPC的情形。下面章节 4 有描述。
· content_type:表示消息data的编码格式名称。实际上RabbitMQ只负责原样传送消息因此不会使用该属性,该属性只被 Publisher Consumer 使用。
消息的确认/删除机制:
Consumer 处理消息可能会失败,那么 RabbitMQ 怎么知道什么时候来删除 queue 中的消息呢?它使用两种机制:
· RabbitMQ 主动将消息发给 Consumer 以后,它会删除消息
· Consumer 发回一个确认后,RabbitMQ 会删除消息。
第二种情况下,如果 RabbitMQ 没收到确认,它会把消息重新放进队列(re-queued)并添加标识 'redelivered' 表明该消息之前已经发送过 ,如果没有Consumer的话,消息将保持到有下一个 Consumer 为止。
Consumer 可以主动告诉 RabbitMQ 消息处理失败了(拒绝消息),并告知RabbitMQ 是删除消息还是重新放进队列。
1.3 exchange 交换机
exchange 有几个重要的属性:
· Name 名字:交换机名字。空字符串名字的exchange为默认的exchange
· Type 类型:Direct, Fanout, Topic, Headers。类型决定 exchange 的消息转发能力。下面 章节2 有描述。
· durable:值为 True/False。值为 true exchange rabbitmq 重启后会被自动创建。OpenStack 使用的exchange的该值都为false
· auto_delete:值为 True/False。设置为 true 的话,当所有消费者的连接都关闭后,该 exchange 会被自动删除。OpenStack 使用的exchange的该值都为false
· exclusive:值为 True/False。设置为 true 的话,该 exchange 只允许被创建的connection使用,并且在该 connection 关闭后它会被自动删除。
RabbitMQ 默认会为每一种类型生成一个或者两个的默认的 exchange
· Fanout 类型:名字为 amq.fanout
· Topic 类型: 名字为 amq.topic
· Headers 类型:名字为 amq.match amq.headers
· Direct 类型:名字为空字符串的exchange 以及 amq.direct。其中名字为空的exchange比较特殊。在一个 Queue 被创建后,RabbitMQ 会自动建立它和该 exchange 之间的binding,并且设置其 binding_key 为该queue 的名字。这样,该语句 channel.basic_publish(exchange='', routing_key='hello',             body=message) 会让该默认的 exchange 将该 message 转发到名字为 'hello' 的队列中。
1.4 队列 Queue
队列同样有类似于 exchange namedurableauto_delete exclusive 等属性,并且含义相同。
Exchange 会将消息分发(copy)到符合要求的所有队列中。
Consumer 可以主动获取或者被动接受Queue里面的消息:
· 被动接收消息(订阅消息 "push API"):使用 basic.consume(shortreserved-1, queue-namequeue, consumer-tagconsumer-tag,no-localno-local, no-ackno-ack, bitexclusive, no-waitno-wait,tablearguments)
方法。见 5.1 示例代码。
· 主动获取消息 ("pull API": 使用 basic.get(shortreserved-1, queue-namequeue, no-ackno-ack) 方法。
1.5 rabbitmqctl  Cli
RabbitMQ 提供Cli  rabbitmqctl [-n <node>] [-q] <command> [<command options>] 来进行管理和配置。常用到的命令有:
· stop/start_app
· add/delete/list_vhosts
· list_queues/exchanges/bindings/connections/channels
· trace_on/off
2 消息转发机制
Exchange 根据它自身的类型 type、消息的属性 routing_key 或者 headers,以及 Binding 的属性 binding_key 来转发消息。
Exchange 的类型 Type
使用的消息属性
使用的Binding 属性
转发模式
Fanout
- (忽略消息的转发属性)
- (忽略binding的转发属性)
Exchange 将消息转发到所有与它有 binding 关系的队列中。
这种方法转发效率较高。OpenStack 大量使用这种类型的 exchange
Direct
routing_key (任意的字符串,比如 "abc"
binding_key (任意的字符串,比如 "abc"
Exchange 只将消息转到 binding binding_key 等于消息的 routing_key 的队列中。
Topic
routing_key (以 "." 分割的多单词字符串,比如 abc.efg.hij
binding_key (包含 "#" "*" 的以 “.” 分割的多单词字符串,比如 *.efg.*
Exchange 只将消息转到消息的 routing_key binding binding_key 匹配的队列中。匹配规则如下:
1)两者以"."分割的单词数目相同
2"*"可代表一个单词
3"#“可代表零个或多个单词
Headers
headers (消息头)
binding_key
Exchange 只将消息转到消息的 headers binding binding_key 匹配的队列中。匹配规则待研究。
OpenStack不使用该类型的exchange
参考文档:
https://www.rabbitmq.com/getstarted.html 这里有详细的阐述和示例源代码。
http://www.cnblogs.com/starof/p/4173413.html 这里有官网文档的中文版。
3 持久化
消息的持久化意味着在 RabbitMQ 被重启后,消息依然还在。要实现持久化,得实现几个相关组件的持久化:
1. 交换机的持久化,需要将其 durable 属性设为 truechan.exchange_declare(exchange="sorting_room", type="direct", durable=True, auto_delete=False,)
2. 队列的持久化,需要将其 durable 属性设置为 truechan.queue_declare(queue="po_box", durable=True, exclusive=False, auto_delete=False)
3. 消息的持久化,需要将其 Delivery Mode 属性设置成2 msg.properties["delivery_mode"] = 2
4 RPC
可以使用 RabbitMQ 来实现 RPC 机制,这里说说其实现原理:
DSC0002.png
过程:
1. 客户端 Client 设置消息的 routing key Service 的队列 op_q;设置消息的 reply-to 属性为返回的 response 的目标队列 reponse_q,设置其 correlation_id 为以随机UUID,然后将消息发到 exchange。比如 channel.basic_publish(exchange='', routing_key='op_q', properties=pika.BasicProperties(reply_to = reponse_q, correlation_id = self.corr_id),body=request)
2. Exchange 将消息转发到 Service op_q
3. Service 收到该消息后进行处理,然后将response 发到 exchange,并设置消息的 routing_key 为原消息的 reply_to 属性,以及设置其 correlation_id 为原消息的 correlation_id
ch.basic_publish(exchange='', routing_key=props.reply_to, properties=pika.BasicProperties(correlation_id = props.correlation_id), body=str(response))
4. Exchange 将消息转发到 reponse_q
5. Client 逐一接受 response_q 中的消息,检查消息的 correlation_id 是否为等于它发出的消息的correlation_id,是的话表明该消息为它需要的response
这里有详细的阐述。
5 Python AMQP SDK
常用的Python AMQP SDK包括:
· py-amqplib (支持 AMQP 0.8): http://barryp.org/software/py-amqplib/
· pika (支持 AMQP 0.9.1Python 2.6 2.7):https://github.com/pika/pika
· txamqp: https://launchpad.net/txamqp
5.1 一个简单的使用 py-amqplib Consumer 实现


#创建Connection和Channel连接到 RabbitMQ 服务器
conn = amqp.Connection(host="localhost:5672", userid="guest", password="1111", virtual_host="/", insist=False)
chan = conn.channel()
#创建 queue
result = chan.queue_declare(queue="debug", durable=True, exclusive=False, auto_delete=False)
#创建 exchange
result = chan.exchange_declare(exchange="sorting_room2", type="topic", durable=True, auto_delete=False,)
#创建 binding
result = chan.queue_bind(queue="debug", exchange="sorting_room2", routing_key="*.debug")
#回调函数,当有 message 到达 queue 后,该函数会被调用def recv_callback(msg):
    print 'Received: ' + msg.body + ' from channel #' + str(msg.channel.channel_id)
    # lChannel.basic_ack(msg.delivery_tag) #如果no_ack=False的话,可以需要发回一个确认
#启动一个 consumer,consumer_tag 是该 consumer 的一个唯一标识符#no_ack = True 表示该 consumer 不会发回确认chan.basic_consume(queue='debug', no_ack=True, callback=recv_callback, consumer_tag="debugtag")
#等待有消息发到 queuewhile True:
    chan.wait()
#终止该 consumer
chan.basic_cancel("testtag")
#关闭 connection 和 channelchan.close()
conn.close()


5.2 一个简单的使用 py-amqplib Producer 实现代码


from amqplib import client_0_8 as amqpimport sys
#创建 connection 和 channel
conn = amqp.Connection(host="localhost:5672", userid="guest", password="1111", virtual_host="/", insist=False)
chan = conn.channel()
#创建 message
msg = amqp.Message(sys.argv[1])
msg.properties["delivery_mode"] = 2
#发送 message
chan.basic_publish(msg,exchange="sorting_room2",routing_key=(sys.argv[2]))
#关闭 connection 和 channelchan.close()
conn.close()


6 插件和消息追踪
RabbitMQ 支持使用插件来支持 ManagementFederationShovel  STOMP。所有的插件都在这里
二、安装rabbitmq
1controller 节点
  安装消息队列服务
  # apt-get install rabbitmq-server -y
DSC0003.png

  新建用户名为openstack密码为51elab的用户:
  #  rabbitmqctl add_user openstack 51elab
  配置用户权限:
  # rabbitmqctl set_user_tags openstack administrator (给用户赋予角色)
  # rabbitmqctl set_permissions -p /  openstack ".*" ".*" ".*" (给予该用户所有权限)
DSC0004.png

  重启服务即可:
  # service rabbitmq-server restart
DSC0005.png

三、插件和消息追踪
#打开监听web插件:
rabbitmq-plugins enable rabbitmq_management
DSC0006.png
#再次重启服务即可
它的GUI的访问地址是 http://192.168.1.141:15672进去后输入之前创建的用户名和密码就可以登录进去了,它的GUI上,提供了一个 overview,还可以通过它来管理connectionchannelexchange queue,以及 virtual hosttracing policy等。

DSC0007.png
DSC0008.png
#Rabbitmq的firehost机制:
该机制提供了一个查看被转发的消息的途径。当打开 firehose 的时候,RabbitMQ 会自动建立 amq.rabbitmq.trace amq.rabbitmq.log 两个exchange。你可以编程创建queue 从这两个 exchange 里面获取 trace log,从而观察每一个被处理的消息。这里有一个开源代码实现
rabbitmq-plugins enable rabbitmq_tracing 打开插件
DSC0009.png
#再次重启服务即可
此时前往 RabbitMQ 管理界面,可以看到在 Admin 标签之下多出了一个栏目 “Tracing”,之前是没有这项的。
DSC00010.png
#消息记录开关:
单纯只是打开 firehose 并不能就获取到消息的记录,由于记录 RabbitMQ 中传递的消息将会影响于性能。所以默认 RabbitMQ 不会打开该功能。要打开,需要执行命令:
rabbitmqctl trace_on
同样,如果不再使用消息历史记录,使用如下命令关闭:
rabbitmqctl trace_off
#查看消息记录:
在 RabbitMQ 管理界面 Admin 标签下,打开 Tracing 栏目。要查看消息需要新建一个 Trace,点击”Add a new trace”,设置一个名称 “all”,表达式中设置 “#”,点击添加。
DSC00011.png
表达式 “#” 意为把所有的消息都记录下来,你可以设置只记录某一个队列或交换机上的消息,如 “#.myqueue”,或者 “#.myexchange”添加之后,可以在 All traces 栏目下,看到新建的 trace。右侧为消息历史记录的文件,打开即可看到 RabbitMQ 中传递的消息历史记录。



  

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表,该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有,请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意,请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息,一经发现立即删除。若您因此触犯法律,一切后果自负,我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责,亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏,不得用于商业或者其他用途,否则,一切后果均由您自己承担,我们对此不承担任何法律责任
7、如涉及侵犯版权等问题,请您及时通知我们,我们将立即采取措施予以解决
8、联系人Email:admin@iyunv.com 网址:www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址:https://www.yunweiku.com/thread-496762-1-1.html 上篇帖子: 关于openstack的NTP安装 下篇帖子: 使用kvm 制作openstack windows镜像
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群,最新一手资源尽在官方微信交流群!快快加入我们吧...

扫描微信二维码查看详情

客服E-mail:kefu@iyunv.com 客服QQ:1061981298


QQ群⑦:运维网交流群⑦ QQ群⑧:运维网交流群⑧ k8s群:运维网kubernetes交流群


提醒:禁止发布任何违反国家法律、法规的言论与图片等内容;本站内容均来自个人观点与网络等信息,非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来,其版权均归原作者及其网站所有,我们尊重他人的合法权益,如有内容侵犯您的合法权益,请及时与我们联系进行核实删除!



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表