Python -多处理，给每个处理器一封来自文本文件的电子邮件。

Question

因此，我一直在玩多重处理，我正在考虑提高我的知识，使我可以从进程1的文本文件中读取第一句，然后从进程2的第二句，等等……

txt file:

helloworld@world.com
helloworld2@world.com
helloworld3@world.com
helloworld4@world.com
helloworld5@world.com

代码就是这样的：

def info(thread):
    global prod
    prod = int(thread) + 1
    runit()


def runit():

    log("Profile-" + str(prod) + Fore.GREEN + ' - ' + email)
    #From here I can then use the email for each worker basically. Or thats the plan atleast. Theplan is that every worker will have its own email that can be used in here.
    sys.exit()

def main():
    user_input = 0
    while True:
        try:
            user_input = int(input(Fore.WHITE + 'How many tasks do you wanna run? [NUMBERS] \n' + Fore.RESET))
        except ValueError:
            print(Fore.RED + "Stop being stupid" + Fore.RESET)
            continue
        else:
            with open('email.txt') as f:
                content = f.readlines()
            content = [x.strip('\n') for x in content]

            try:
                for i, email in enumerate(content):
                    print(email)

            except ValueError as e:
                print(e)

            HowManyThread = user_input
            i = 0
            jobs = []
            for i in range(HowManyThread):
                p = multiprocessing.Process(target=info, args=(str(i),))
                jobs.append(p)
                time.sleep(.5)
                p.start()

            for p in jobs:
                p.join()

            sys.exit()

日志基本上只是一条日志消息，没有什么特别的。

Fore.COLOR <--颜色

然而，我完全不知道我应该怎么做，使每一个进程采取每一个电子邮件行。所以基本上..。

Process-1 to take helloworld@world.com
Process-2 to take helloworld2@world.com
Process-3 to take helloworld3@world.com
Process-4 to take helloworld4@world.com
Process-5 to take helloworld5@world.com

对于我如何做到这一点，有什么建议？我完全离开了，完全不知道该如何前进。

更新

from multiprocessing import pool, Process, Queue
from tqdm import tqdm

with open('email.txt') as f:
    content = f.readlines()

global email_list
email_list = [x.strip('\n') for x in content]


def info(thread):
    global prod
    prod = int(thread) + 1
    runit()


def runit(email_index):
    email = email_list[email_index]

    log("Profile-" + str(prod) + Fore.GREEN + ' - ' + email)
    sys.exit()



def main():
    wipe()
    text()
    Infotext = "First name : Last name : Email: : Random char + Street"
    with open('data.json', 'w') as f:
        json.dump(Infotext, f)
        f.write("\n")

    with Pool(8) as pool:
        result_list = list(tqdm(pool.imap_unordered(, range(len(email_list)), chunksize=5), total=len(email_list))))


if __name__ == '__main__':
    try:
        main()

    except Exception as e:
        print(e)
        print(traceback.print_exc())
        print(traceback)

Answer 1

下面的方法将多处理委托给一个工作人员池，每个工人池接收一组索引，并一次处理这些索引(这里的poolsize=8和chunksize=5的选择是任意的，可以根据您的需求进行调整)。

然后，将所有工人的结果汇总到最后的列表中。请注意，只有当您不关心处理行的顺序时，imap_unordered才是合适的(也就是说，result_list不保持content的原始顺序。

from multiprocessing import Pool
# progress bar to track your multiproc
from tqdm import tqdm

with open('email.txt') as f:
    content = f.readlines()

# this list will be accessed by each worker
global email_list
email_list = [x.strip('\n') for x in content]

# define function that worker will apply to each email
# it gets sent an index for the list of emails
# it accesses the email at that index, performs its function and returns
def runit(email_index):
    email = email_list[email_index]
    # do the stuff you're interested in for a single email

# run the multiprocessing to get your results
# this sends the indexes for the emails out to the workers
# and collects the results of runit into result list
with Pool(8) as pool:                                                
  result_list = list(tqdm(pool.imap_unordered(runit,    
                          range(len(email_list)), chunksize=5),                 
                          total=len(email_list)))

Answer 2

您需要的是一个工作进程池--即使对于用例，我也很想知道线程(或multiprocessing.dummy)是否是不够的。

池启动指定数量的工作进程，您可以将异步任务提交到将由第一个空闲工作处理的池中。

简化后的示例版本(没有花哨的打印，没有对列表中的顺序文件进行不必要的读取)可以是：

import multiprocessing
import time

def runit(prod, email):

    print("Profile-" + str(prod) + ' - ' + email)
    #From here I can then use the email for each worker basically. Or thats the plan atleast. Theplan is that every worker will have its own email that can be used in here.
    # sys.exit() # NEVER CALL EXPLICITELY sys.exit() in a worker process
    time.sleep(1) # to add a delay inside each task

def main():
    while True:
        try:
            HowManyThread = int(input(
                'How many tasks do you wanna run? [NUMBERS] \n'))
        except ValueError:
            print("Stop being stupid")
            continue

        if HowManyThread == 0: break

        pool = multiprocessing.Pool(HowManyThread)
        with open('email.txt') as f:
            for i, email in enumerate(f):
                email = email.strip()
                # runit will be runned by a worker process
                pool.apply_async(runit, (i, email))

        pool.close()         # no more task to add
        pool.join()          # wait for last worker to end

if __name__ == "__main__":
    main()