Python - 在独立线程中运行的函数之间传递函数（回调）变量

问题描述：

我正在尝试开发一个使用pika和线程模块的Python 3.6脚本。Python - 在独立线程中运行的函数之间传递函数（回调）变量

我有一个问题，我认为是由我的A）是非常新的Python和编码一般，和B）我不理解如何在函数之间传递变量时，他们在单独的线程中运行，并已通过接收函数名称末尾的括号中的参数。

我认为这是因为当我不使用线程时，我可以简单地通过调用接收函数名称并提供要传递的变量，在括号中显示一个基本示例如下：

def send_variable(): 
    body = "this is a text string" 
    receive_variable(body) 

def receive_variable(body): 
    print(body)

这在运行时，打印：

this is a text string

代码的工作版本，我需要得到具有线程工作如下所示 - 本采用直板功能（无螺纹）和我我使用pika接收来自（RabbitMQ）q的消息ueue通过鼠兔回调函数，然后我在“回调”功能接收消息的身体传递给“处理功能”：

import pika 
...mq connection variables set here... 


# defines username and password credentials as variables set at the top of this script 
    credentials = pika.PlainCredentials(mq_user_name, mq_pass_word) 

# defines mq server host, port and user credentials and creates a connection 
    connection = pika.BlockingConnection(pika.ConnectionParameters(host=mq_host, port=mq_port, credentials=credentials)) 

# creates a channel connection instance using the above settings 
    channel = connection.channel() 

# defines the queue name to be used with the above channel connection instance 
    channel.queue_declare(queue=mq_queue) 


def callback(ch, method, properties, body): 

# passes (body) to processing function 
    body_processing(body) 

# sets channel consume type, also sets queue name/message acknowledge settings based on variables set at top of script 
    channel.basic_consume(callback, queue=mq_queue, no_ack=mq_no_ack) 
# tells the callback function to start consuming 
    channel.start_consuming() 
# calls the callback function to start receiving messages from mq server 
    callback() 
# above deals with pika connection and the main callback function 


def body_processing(body): 
    ...code to send a pika message every time a 'body' message is received...

这工作得很好，但是我希望把这种在脚本中运行使用线程。当我这样做时，我必须提供参数'channel'给在其自己的线程中运行的函数名 - 当我然后尝试包含'body'参数以便'processing_function'看起来如下：

def processing_function(channel, body):

我得到一个错误说：

[function_name] is missing 1 positional argument: 'body'

我知道，使用线程时，有需要更多的代码，我已经列入我用下面线程的实际代码，这样你可以看到我在做什么：

...imports and mq variables and pika connection details are set here... 

def get_heartbeats(channel): 
channel.queue_declare(queue=queue1) 
#print (' [*] Waiting for messages. To exit press CTRL+C') 

    def callback(ch, method, properties, body): 

     process_body(body) 

     #print (" Received %s" % (body)) 

    channel.basic_consume(callback, queue=queue1, no_ack=no_ack) 
    channel.start_consuming() 


def process_body(channel, body): 
    channel.queue_declare(queue=queue2) 
    #print (' [*] Waiting for Tick messages. To exit press CTRL+C') 

# sets the mq host which pika client will use to send a message to 
    connection = pika.BlockingConnection(pika.ConnectionParameters(host=mq_host)) 
# create a channel connection instance 
    channel = connection.channel() 
# declare a queue to be used by the channel connection instance 
    channel.queue_declare(queue=order_send_queue) 
# send a message via the above channel connection settings 
    channel.basic_publish(exchange='', routing_key=send_queue, body='Test Message') 
# send a message via the above channel settings 
# close the channel connection instance 
    connection.close() 


def manager(): 

# Channel 1 Connection Details - ======================================================================================= 

    credentials = pika.PlainCredentials(mq_user_name, mq_password) 
    connection1 = pika.BlockingConnection(pika.ConnectionParameters(host=mq_host, credentials=credentials)) 
    channel1 = connection1.channel() 

# Channel 1 thread ===================================================================================================== 
    t1 = threading.Thread(target=get_heartbeats, args=(channel1,)) 
    t1.daemon = True 
    threads.append(t1) 
    # as this is thread 1 call to start threading is made at start threading section 

# Channel 2 Connection Details - ======================================================================================= 

    credentials = pika.PlainCredentials(mq_user_name, mq_password) 
    connection2 = pika.BlockingConnection(pika.ConnectionParameters(host=mq_host, credentials=credentials)) 
    channel2 = connection2.channel() 

# Channel 2 thread ==================================================================================================== 
    t2 = threading.Thread(target=process_body, args=(channel2, body)) 
    t2.daemon = True 
    threads.append(t2) 
    t2.start() # as this is thread 2 - we need to start the thread here 

# Start threading 
t1.start() # start the first thread - other threads will self start as they call t1.start() in their code block 
for t in threads: # for all the threads defined 
    t.join() # join defined threads 

manager() # run the manager module which starts threads that call each module

这在运行时产生错误

process_body() missing 1 required positional argument: (body)

，我不明白这是为什么或如何解决它。

感谢您花时间阅读此问题，并且您可以提供任何帮助或建议，我们对此表示赞赏。

请记住，我是python和编码的新手，所以可能需要拼写出来的东西，而不是能够理解更加神秘的回复。

谢谢！

答

关于进一步寻找到这一点，并用它看来，如果我编辑的行代码播放：

def process_body(channel, body):

阅读

def process_body(body):

和

t2 = threading.Thread(target=process_body, args=(channel2, body))

使它读取：

t2 = threading.Thread(target=process_body)

那么代码似乎需要工作 - 我也看到了HTOP多个脚本流程，这样看来，穿线工作 - 我已经离开了脚本处理24小时+并没有收到任何错误...

Python - 在独立线程中运行的函数之间传递函数（回调）变量

相关推荐