消息队列_RabbitMQ-0003.深入RabbitMQ节点/配置/管理及日志实时化?

Posted 2020-08-21

理解节点:

说明: 节点是指Erlang节点,而且节点之间支持相互通信,RabbitMQ应用跑在Erlang节点之上,应用崩溃,Erlang节点会自动尝试重启应用程序,前提是Erlang本身没有崩溃,节点日志默认位于var/log/rabbitmq/[email protected]log

基本管理:

---

启动节点: rabbitmq-server -detached

关闭节点: rabbitmqctl stop -n [email protected]

---

说明: rabbitmq-server启动后会自动启动Erlang节点之上的所有应用,新版本前台运行时ctrl+c时会自动停止整个节点而不在询问,rabbitmqctl作为几乎所有的RabbitMQ管理的一站式解决方案,所以更推荐使用stop优雅关闭整个节点,节点一旦关闭,上层的应用程序自然也会全部关闭.

配置文件:

---

etc/rabbitmq/rabbitmq-env.conf

说明: 此配置文件主要用于定义节点环境变量,具体可参考(http://www.rabbitmq.com/configure.html#define-environment-variables)

etc/rabbitmq/rabbitmq.config

说明: 此匹配文件主要用于定义节点组件配置,具体可参考(http://www.rabbitmq.com/configure.html#configuration-file)

应用管理:

---

关闭应用: rabbitmqctl stop_app

开启应用: rabbitmqctl start_app

---

注意: stop_app只会停止应用程序而不会停止Erlang节点,常用于集群加入之前停止应用程序,把节点重置为原始状态

用户管理:

添加用户: rabbitmqctl add_user root qwertyuiop

删除用户: rabbitmqctl delete_user root

列出用户: rabbitmqctl list_users

更改密码: rabbitmqctl change_password root qwertyuiop

验证密码: rabbitmqctl authenticate_user root qwertyuiop

权限管理:

设置权限: rabbitmqctl set_permissions -p / root ".*" ".*" ".*"

更改权限: rabbitmqctl set_permissions -p / root "" "salt-.*" ".*"

清除权限: rabbitmqctl clear_permissions -p / root

列出权限: rabbitmqctl list_permissions

用户权限: rabbitmqctl list_user_permissions root

---

说明: 访问控制条目包含目标Vhost,目标帐号,配置权限,写权限,读权限,如上-p后必须跟目标Vhost,后面的三个正则表达式参数".*",".*",".*"分别表示配置权限正则,写权限正则,读权限正则,"" "salt-.*" ".*"表示完全阻止配置操作,允许对salt-开头的队列和交换机写,对所有的队列和交换机可读.

案例驱动:

说明: 为了演示RabbitMQ交换机/队列/绑定管理操作,按照如上图创建一个主题交换机绑定三个队列分别用于接收所有错误日志(绑定路由键*.errs),所有日志(绑定路由键alls.*),正常日志(绑定路由键*.info)

---

> 生产者

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# Purpose:
#
"""
# 说明: 导入公共模块
import sys
import pika
# 说明: 导入其它模块


if __name__ == ‘__main__‘:
    # 创建凭证对象
    credentials = pika.PlainCredentials(‘root‘, ‘qwertyuiop‘)
    # 创建参数对象
    conn_params = pika.ConnectionParameters(
        # RabbitMQ服务地址
        host=‘127.0.0.1‘,
        # RabbitMQ服务端口
        port=5672,
        # RabbitMQ登录凭证
        credentials=credentials,
        # RabbitMQ虚拟主机
        virtual_host=‘/‘
    )
    # 创建连接对象
    conn_broker = pika.BlockingConnection(conn_params)
    # 获取信道对象
    channel = conn_broker.channel()
    exchange = sys.argv[1]
    routekey = sys.argv[2]
    messages = sys.argv[3]
    # 创建配置对象
    msg_props = pika.BasicProperties()
    # 设置内容类型
    msg_props.content_type = ‘text/plain‘
    # 尝试发布消息
    channel.basic_publish(
        # 发布消息内容
        body=messages,
        # 发布到交换机
        exchange=exchange,
        # 发布信息属性
        properties=msg_props,
        # 发布信息时携带的路由键
        routing_key=routekey
    )

---

> 消费者

---

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# Purpose:
#
"""
# 说明: 导入公共模块
import pika
# 说明: 导入其它模块
if __name__ == ‘__main__‘:
    # 创建凭证对象
    credentials = pika.PlainCredentials(‘root‘, ‘qwertyuiop‘)
    # 创建参数对象
    conn_params = pika.ConnectionParameters(
        # RabbitMQ服务地址
        host=‘127.0.0.1‘,
        # RabbitMQ服务端口
        port=5672,
        # RabbitMQ登录凭证
        credentials=credentials,
        # RabbitMQ虚拟主机
        virtual_host=‘/‘
    )
    # 创建连接对象
    conn_broker = pika.BlockingConnection(conn_params)
    # 获取信道对象
    channel = conn_broker.channel()
    # 创建日志交换机
    channel.exchange_declare(
        # 交换机名称
        exchange="logs-exchange",
        # 交换机类型
        type="topic",
        # 如果同名交换机已存在依然返回成功,否则创建
        passive=False,
        # 声明为持久化交换机
        durable=True,
        # 交换机闲置也不会自动删除
        auto_delete=False
    )
    # 创建info日志级别队列
    channel.queue_declare(
            # 队列名称
            queue="info",
            # 如果同名队列已存在依然返回成功,否则创建
            passive=False,
            # 声明为持久化队列
            durable=True,
            # 声明为非私有队列
            exclusive=False,
            # 队列闲置也不会自动删除
            auto_delete=False
    )
    # 创建errs日志级别队列
    channel.queue_declare(
            # 队列名称
            queue="errs",
            # 如果同名队列已存在依然返回成功,否则创建
            passive=False,
            # 声明为持久化队列
            durable=True,
            # 声明为非私有队列
            exclusive=False,
            # 队列闲置也不会自动删除
            auto_delete=False
    )
    # 创建alls日志级别队列
    channel.queue_declare(
            # 队列名称
            queue="alls",
            # 如果同名队列已存在依然返回成功,否则创建
            passive=False,
            # 声明为持久化队列
            durable=True,
            # 声明为非私有队列
            exclusive=False,
            # 队列闲置也不会自动删除
            auto_delete=False
    )
    # 绑定队列
    channel.queue_bind(
        # 队列名称
        queue="info",
        # 交换机名称
        exchange="logs-exchange",
        # 路由键名称
        routing_key="*.info"
    )
    channel.queue_bind(
        # 队列名称
        queue="errs",
        # 交换机名称
        exchange="logs-exchange",
        # 路由键名称
        routing_key="*.errs"
    )
    def msg_consumer(channel, method, header, body):
        # 发送消息确认
        channel.basic_ack(delivery_tag=method.delivery_tag)
        print ‘#[{0}]>: {1}‘.format(method.delivery_tag, body)
        return
    # 作为指定队列消费者
    channel.basic_consume(msg_consumer, queue="info", consumer_tag="logs-exchange-info")
    channel.basic_consume(msg_consumer, queue="errs", consumer_tag="logs-exchange-errs")
    channel.basic_consume(msg_consumer, queue="alls", consumer_tag="logs-exchange-alls")
    # 循环调用回调函数接收处理消息
    channel.start_consuming()
    channel.close()

说明: 测试方法非常简单,python consumer.py启动后尝试执行python producer.py logs-exchange msg.info/python producer.py logs-exchange msg.errs/python producer.py logs-exchange alls.errs关注消费者那边的变化~

---

队列管理:

列出队列: rabbitmqctl list_queues -p / name messages

列出队列: rabbitmqadmin  -V / list queues name messages

删除队列: rabbitmqctl eval ‘IfUnused = false, IfEmpty = true, MatchRegex = <<"info">>, [rabbit_amqqueue:delete(Q, IfUnused, IfEmpty) || Q <- rabbit_amqqueue:list(), re:run(element(4, element(2, Q)), MatchRegex) =/= nomatch ].‘

删除队列: rabbitmqadmin   -V / delete queue name=info

---

说明: list_queues默认只会列出name,messages列,如果要列出更多可以空格隔开<queueinfoitem>,通过rabbitmqctl eval删除非常不方便,具体参数可以修改来删除指定的队列,当然也可以使用rabbitmqadmin调用api删除非常方便便于脚本集成,但必须开启rabbitmq_management管理.

交换管理:

列出交换: rabbitmqctl -p / list_exchanges name type

列出交换: rabbitmqadmin  -V / list exchanges name type

列出绑定: rabbitmqctl -p / list_bindings

列出绑定: rabbitmqadmin -V / list bindings

删除交换: rabbitmqadmin -V / delete exchange name=logs-exchange

---

说明: list_exchanges默认只会列出name,type列,如果要列出更多可以空格隔开<exchangeinfoitem>,使用rabbitmqadmin调用api删除非常方便便于脚本集成,但必须开启rabbitmq_management管理.

插件管理:

启用插件: rabbitmq-plugins enable rabbitmq_management

关闭插件: rabbitmq-plugins disable rabbitmq_management

统计数据:

状态统计: rabbitmqctl status

还原配置:

恢复出厂: rabbitmqctl stop_app && rabbitmqctl reset && rabbitmqctl start_app

日志管理:

文本读取: var/log/rabbitmq/[email protected]

文本读取: var/log/rabbitmq/[email protected].log

日志轮询: rabbitmqctl rotate_logs .suffix

实时日志: rabbitmqadmin list exchanges查看默认交换机中有一个amq.rabbitmq.log交换机,下面代码伪造了三个零时队列分别接收routing_key为info/warning/error级别的实时日志,非常简单~

---

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# Purpose:
#
"""
# 说明: 导入公共模块
import pika
# 说明: 导入其它模块
if __name__ == ‘__main__‘:
    # 创建凭证对象
    credentials = pika.PlainCredentials(‘guest‘, ‘guest‘)
    # 创建参数对象
    conn_params = pika.ConnectionParameters(
        # RabbitMQ服务地址
        host=‘127.0.0.1‘,
        # RabbitMQ服务端口
        port=5672,
        # RabbitMQ服务凭证
        credentials=credentials,
        # RabbitMQ虚拟主机
        virtual_host=‘/‘
    )
    # 创建连接对象
    conn_broker = pika.BlockingConnection(conn_params)
    # 获取信道对象
    channel = conn_broker.channel()
    # 创建队列
    channel.queue_declare(
            # 队列名称
            queue="amq_info",
            # 如果同名队列已存在依然返回成功,否则创建
            passive=False,
            # 声明为非持久化队列
            durable=False,
            # 声明为私有队列
            exclusive=True,
            # 队列闲置会自动删除
            auto_delete=True
    )
    channel.queue_declare(
            # 队列名称
            queue="amq_warn",
            # 如果同名队列已存在依然返回成功,否则创建
            passive=False,
            # 声明为非持久化队列
            durable=False,
            # 声明为私有队列
            exclusive=True,
            # 队列闲置会自动删除
            auto_delete=True
    )
    channel.queue_declare(
            # 队列名称
            queue="amq_errs",
            # 如果同名队列已存在依然返回成功,否则创建
            passive=False,
            # 声明为非持久化队列
            durable=False,
            # 声明为私有队列
            exclusive=True,
            # 队列闲置会自动删除
            auto_delete=True
    )
    # 绑定队列
    channel.queue_bind(
        # 队列名称
        queue="amq_info",
        # 交换机名称
        exchange="amq.rabbitmq.log",
        # 路由键名称
        routing_key="info"
    )
    # 绑定队列
    channel.queue_bind(
        # 队列名称
        queue="amq_warn",
        # 交换机名称
        exchange="amq.rabbitmq.log",
        # 路由键名称
        routing_key="warning"
    )
    # 绑定队列
    channel.queue_bind(
        # 队列名称
        queue="amq_errs",
        # 交换机名称
        exchange="amq.rabbitmq.log",
        # 路由键名称
        routing_key="error"
    )
    # 消息回调处理函数
    def msg_consumer(channel, method, header, body):
        # 发送消息确认
        channel.basic_ack(delivery_tag=method.delivery_tag)
        print ‘#[{0}]>: {1}‘.format(method.delivery_tag, body)
        return
    # 作为指定队列消费者
    channel.basic_consume(msg_consumer, queue="amq_info", consumer_tag="amq.rabbitmq.log.info")
    channel.basic_consume(msg_consumer, queue="amq_warn", consumer_tag="amq.rabbitmq.log.warn")
    channel.basic_consume(msg_consumer, queue="amq_errs", consumer_tag="amq.rabbitmq.log.errs")
    # 循环调用回调函数接收处理消息
    channel.start_consuming()
    channel.close()

---

说明: [email protected]中常常会记录Erlang崩溃或错误原因,当节点无法启动时候对调试非常有用,[email protected].log会记录生产者和消费者的所有事件,调试时非常有用,rotate_logs会以.suffix为后缀切割日志,大量日志crontab轮询时非常有用.

本文出自 “满满李 - 运维开发之路” 博客，请务必保留此出处http://xmdevops.blog.51cto.com/11144840/1878270