用于从rabbitmq获取消息的python多线程应用程序

Question

Worker.counter =0 confobj.thread_count =2

def callback(ch, method, properties, body):
    if(Worker.counter<confobj.thread_count):
      logObject = json.loads(body)
      th=Worker(Worker.counter+1,logObject,confobj,fileobj)
      Worker.counter+=1
      th.start()
    else:
      print("All threads are busy.") 
      time.sleep(2)  
    ch.basic_ack(delivery_tag = method.delivery_tag)

  channel.basic_qos(prefetch_count=1)
  channel.basic_consume(callback,
                        queu`enter code here`e=confobj.queue)
  channel.start_consuming()

我希望 maxthread 应该是 2 以便第一个线程访问第一条消息，第二个线程访问第二条消息，第一个线程再次访问第三条消息，依此类推 然后我想将该消息插入弹性搜索而不跳过任何消息。

Answer 1

您可以尝试使用来自 Multiprocessing 的 Process 并调用 Process.pid 函数来获取对每个进程 ID 的访问权限。

Answer 2

from multiprocessing import Process, current_process


def square(x):
print(x*x)
print('processID =', current_process())
return x * x


def cube(y):
print(y*y*y)
print('processID =', current_process())
return y * y * y


if __name__ == '__main__':
number = 6
one = Process(target=square, args=(number,))
two = Process(target=cube, args=(number,))

one.start()
two.start()

one.join()
two.join()

这里是一个简单的例子，使用具有不同功能的 Process。因此，您可以通过进程 ID 调用 .start()。 （这不是一个复杂的代码。这只是一个理解的例子）。