python操作rabbitmq的consumer,并强制关闭特定consumer的连接

引入

python获取指定队列的consumer

python可以使用rabbitmq_management的api获取所有consumer的信息。【队列需要安装rabbitmq_management，在大部分情况下，我们安装队列的同时，都会安装管理界面，以方便一些简单的手动操作与简单监控】

获取consumer详情

以下代码就可以输出指定队列的consumer详情

import requests
from requests.auth import HTTPBasicAuthrabbitmq_host = '10.139.1.124'
# RabbitMQ管理API的URL
url = f'http://{rabbitmq_host}:15672/api/consumers'# RabbitMQ的用户名和密码
username = 'guest'
password = 'guest'response = requests.get(url, auth=HTTPBasicAuth(username, password))# 检查请求是否成功
if response.status_code == 200:consumers = response.json()for consumer in consumers:print(consumer)

**f’http://{rabbitmq_host}:15672/api/consumers’**这个api是我们常用与consumer操作的接口
运行这个代码之后，一般会输出多行的consumer信息，每条信息如下：

{'arguments': {}, 'ack_required': True, 'active': True, 'activity_status': 'up', 'channel_details': {'connection_name': '10.139.1.124:60440 -> 172.17.0.4:5672', 'name': '10.139.1.124:60440 -> 172.17.0.4:5672 (1)', 'node': 'rabbit@31ae4d091d79', 'number': 1, 'peer_host': '10.139.1.124', 'peer_port': 60440, 'user': 'guest'}, 'consumer_tag': 'ctag1.c48bbeb0d7d241519fd54553bc73c1b8', 'exclusive': False, 'prefetch_count': 1, 'queue': {'name': 'amap_work_error', 'vhost': '/'}}

这些信息包含了消费者的状态，消费者id，消费的队列信息，确认消费信息等。详细的字段解释如下：

consumer详情字段解释

{'arguments': {},  # 消费者的额外参数'ack_required': True,  # 是否需要确认消息'active': True,  # 消费者是否活跃'activity_status': 'up',  # 消费者的活动状态'channel_details': {  # 消费者的通道详情'connection_name': '10.139.1.124:60440 -> 172.17.0.4:5672',  # 连接名称'name': '10.139.1.124:60440 -> 172.17.0.4:5672 (1)',  # 通道名称'node': 'rabbit@31ae4d091d79',  # RabbitMQ节点名称'number': 1,  # 通道编号'peer_host': '10.139.1.124',  # 连接的主机'peer_port': 60440,  # 连接的端口'user': 'guest'  # 用户名},'consumer_tag': 'ctag1.c48bbeb0d7d241519fd54553bc73c1b8',  # 消费者标签'exclusive': False,  # 消费者是否为独占'prefetch_count': 1,  # 预取消息数量'queue': {'name': 'amap_work_error',  # 队列名称'vhost': '/'  # 虚拟主机}
}

这也我们只需要对consumer信息内略加过滤就可以满足自己的大部分需求

强制关闭consumer【通过关闭rabbitmq的通道实现】

cookies = {'_ga': 'GA1.1.1271813516.1712559218','_ga_4NJG4BH26D': 'GS1.1.1713228409.2.0.1713228409.0.0.0','Hm_lvt_bd78bc908e66174e7dde385bf37cb4c1': '1714963778,1715568623,1715824211,1716187398','Hm_lpvt_bd78bc908e66174e7dde385bf37cb4c1': '1716519695','m': '2258:Z3Vlc3Q6Z3Vlc3Q%253D',
}headers = {'Accept': '*/*','Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8','Cache-Control': 'no-cache','Origin': 'http://10.139.1.132:15672','Pragma': 'no-cache','Proxy-Connection': 'keep-alive','Referer': 'http://10.139.1.132:15672/','User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36','X-Reason': 'Closed via management plugin','authorization': 'Basic Z3Vlc3Q6Z3Vlc3Q=','content-type': 'application/json',
}source = "10.139.1.127:53199 -> 172.17.0.3:5672"
encoded_url = quote(source, safe='')close_demo_url = f'http://10.139.1.132:15672/api/connections/{encoded_url}'
print(close_demo_url)
json_data = {"name": "10.139.1.127:53199 -> 172.17.0.3:5672","reason": "Closed via management plugin"
}
response = requests.delete(url,cookies=cookies,headers=headers,json=json_data,verify=False, auth=HTTPBasicAuth(username, password))
print(response.status_code)