Python中如何将脚本1从RabbitMQ接收的值传递给脚本2？

• 脚本1的__init__方法中调用了self.channel.start_consuming()，这是阻塞方法，会一直停在该步骤等待消息，导致后续receive_mq.call()代码永远无法执行。
• 脚本2被导入时会自动执行底部的httpAuthenticate和httpAddQueueItems调用，此时strQueue还是初始的None，自然无法获取RabbitMQ消息。
• 脚本1中on_response方法里的strQueue = body.decode()只是创建局部变量，并未修改脚本2中的全局strQueue变量。

步骤1：修改脚本2，封装逻辑为可调用函数 #

将脚本2的执行逻辑封装成函数，避免导入时自动执行，等待外部传入消息后再触发：

import requests
import json

# Function Authenticate
def httpAuthenticate(in_apiusers, in_apipass, in_Tenant, in_URL):
    try:
        print('retrieve token...')
        url = in_URL
        payload = json.dumps({
            "password": str(in_apipass),
            "usernameOrEmailAddress": str(in_apiusers),
            "tenancyName": str(in_Tenant)
        })
        headers = {
            'Content-Type': 'application/json'
        }
        response = requests.request("POST", url, headers=headers, data=payload)
        json_object = json.loads(response.text)
        print('token code: ' + str(response.status_code))
        return str(json_object["result"])
    except Exception as e:
        return 'Fail:'


# Function:Add Queue Item on Uipath Orchestrator
def httpAddQueueItems(in_URL, in_Token, in_QueueName, in_strjson):
    try:
        print('add queue item...')
        url = in_URL
        # 修复原代码笔误：in_strpjson改为in_strjson
        payload = json.dumps({
            "itemData": {
                "Priority": "Normal",
                "Name": str(in_QueueName),
                "SpecificContent": {
                    "in_pjsorequest": in_strjson
                },
                "Reference": "ggg"
            }
        })
        headers = {
            'X-UIPATH-OrganizationUnitId': '',
            'Content-Type': 'application/json',
            'Authorization': 'Bearer ' + in_Token
        }
        response = requests.request("POST", url, headers=headers, data=payload)
        print(f"Add queue response status: {response.status_code}")
    except Exception as e:
        print(e)
        return 'Fail'

# 封装主逻辑为函数，接收消息参数
def process_message(message):
    # CONSTANTS
    OnPremuser = "xxxx"
    OnPrempass = "xxx!"
    OnPremtenant = "Default"
    OnPremUrlAuth = "xxxx"
    OnPremUrlAddQueue = "https://xxxx"
    OnPremQueue = "JSON"
    
    OnPremtoken = httpAuthenticate(OnPremuser, OnPrempass, OnPremtenant, OnPremUrlAuth)
    if not OnPremtoken.startswith('Fail:'):
        httpAddQueueItems(OnPremUrlAddQueue, OnPremtoken, OnPremQueue, message)
    else:
        print(f"Authentication failed: {OnPremtoken}")

步骤2：修改脚本1，收到消息后调用脚本2的函数 #

调整脚本1逻辑，移除__init__中的阻塞前置调用，让消息回调直接触发脚本2的处理逻辑：

import pika
import time
from script2 import process_message

class ReceiveFromMQ(object):
    def __init__(self):
        credentials = pika.PlainCredentials('xxxx', 'xxxx')
        parameters = pika.ConnectionParameters('xxxx', xxx, 'xxx', credentials)
        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(
            queue='queue',
            on_message_callback=self.on_response,
            auto_ack=True)
        print(' [*] Waiting for messages. To exit press CTRL+C')
        # 将start_consuming放在最后，确保消费者启动后持续监听消息
        self.channel.start_consuming()

    def on_response(self, ch, method, props, body):
        message = body.decode()
        print(" [x] Received %r" % message)
        # 收到消息后直接调用脚本2的处理函数
        process_message(message)
        time.sleep(body.count(b'.'))
        print(" [x] Done")

# 启动消费者
if __name__ == "__main__":
    ReceiveFromMQ()

关键修改说明 #

1. 脚本2不再在导入时自动执行逻辑，通过process_message函数接收外部传入的消息后，再触发认证和添加队列操作。
2. 脚本1移除多余的call方法，让on_response回调直接处理消息并调用脚本2的函数，确保只有收到消息后才执行脚本2的逻辑。
3. 修复脚本2中httpAddQueueItems函数的参数笔误，避免参数引用错误。
4. 脚本1将start_consuming放在__init__最后，确保消费者启动后持续监听消息，收到消息立即触发处理。

内容的提问来源于stack exchange，提问作者lola