Python中如何实现基于队列的分布式机器学习任务处理

pip install pika

import pika

USERNAME = 'guest'
PASSWORD = 'guest'
HOSTNAME = 'localhost'
PORT = 5672

credentials = pika.PlainCredentials(USERNAME, PASSWORD)
parameters = pika.ConnectionParameters(HOSTNAME, PORT, '/', credentials)

import threading
import time

QUEUE_NAME = 'ml_tasks'


def receive_message(ch, method, properties, body):
    print(f'Received message: {body}')
    # 在这里添加具体的机器学习任务处理逻辑


def consume_queue():
    connection = pika.BlockingConnection(parameters)
    channel = connection.channel()
    channel.queue_declare(queue=QUEUE_NAME)
    channel.basic_consume(queue=QUEUE_NAME, on_message_callback=receive_message, auto_ack=True)
    channel.start_consuming()


def send_message(message):
    connection = pika.BlockingConnection(parameters)
    channel = connection.channel()
    channel.queue_declare(queue=QUEUE_NAME)
    channel.basic_publish(exchange='', routing_key=QUEUE_NAME, body=message)
    print(f'Sent message: {message}')
    connection.close()


# 在一个线程中启动任务消费者
thread = threading.Thread(target=consume_queue)
thread.start()

# 发送一个任务消息
send_message('pidancode.com')

# 等待一段时间，确保任务消息被处理完
time.sleep(1)

# 关闭连接
thread.stop()  # 此方法已废弃，应该使用更安全的方式停止线程
thread.join()