程序代码：

    time.sleep(3)
    queue = Queue()
    New_start = input(('导入批量url文本:'))
    levels = int(input(('设置扫描等级:')))
    countss = int(input(('设置扫描进程数:')))
    p = multiprocessing.Pool(countss)
    list_ = list(set(
        [x.replace('\n', '') if x.startswith('http') else 'http://' + x.replace('\n', '') for x in
         open(New_start, 'r', encoding='UTF-8').readlines()]))
    for i in list_:
        queue.put(i)
        print(i + " -------> 加入队列 当前队列总任务数:%s" % queue.qsize())
    while not queue.empty():
        # 子进程原来跑 进行测试
        p.apply_async(get_url_sql, args=(queue.get(), levels))
    p.close()
    p.join()
    print("当前队列中的任务数为:%s" % queue.qsize())
    while True:
        if queue.qsize() < 10:
            opm = OPMysql()
            sql = "select url from url_index limit 1,20"
            res = opm.op_select(sql)
            _list = [i[j].replace('\r\n', '').replace(' ', '')
                     for i in res for j in i]
            url_list = list(set(_list))
            for i in url_list:
                time.sleep(0.5)
                queue.put(i)
                print(i + " -------> 加入队列 当前队列总任务数:%s" % queue.qsize())
            opm.dispose()  # 释放资源

[此贴子已经被作者于2019-9-26 16:54编辑过]

程序代码：

    while not queue.empty():
        # 子进程原来跑 进行测试
        p.apply_async(get_url_sql, args=(queue.get(), levels))
        p.apply_async(tttttta, args=(queue,))
    p.close()
    p.join()

程序代码：

def tttttta(qqqq):
    print("当前队列中的任务数为:%s" % qqqq.qsize())
    while True:
        if qqqq.qsize() < 10:
            opm = OPMysql()
            sql = "select url from url_index limit 1,20"
            res = opm.op_select(sql)
            _list = [i[j].replace('\r\n', '').replace(' ', '')
                     for i in res for j in i]
            url_list = list(set(_list))
            for i in url_list:
                time.sleep(0.5)
                qqqq.put(i)
                print(i + " -------> 加入队列 当前队列总任务数:%s" % qqqq.qsize())
            opm.dispose()  # 释放资源

程序代码：

# -*- coding:utf-8 -*-
from multiprocessing import Pool, Queue, Manager
import time
import os


def run(q):
    
    print("子进程开始，进程ID：%d" % (os.getpid()))
    while True:
        print("循环一次")
        print(" -------> 当前队列 %d 总任务数:%d" % (os.getpid(),q.qsize()))
        if q.qsize() > 0:
            
            qqqq_info = q.get()
            print(" -------> 队列 %d 获取任务:%s" % (os.getpid(),qqqq_info))
            print(" -------> 当前队列 %d 总任务数:%d" %  (os.getpid(),q.qsize()))
            time.sleep(1)
        else:
            print("队列已经取完,跳出循环,子进程结束")
            break
    print("子进程结束，进程ID：%d。" % (os.getpid()))


if __name__ == "__main__":
    print("父进程开始")
    #创建Queue队列管理Manager,有了Manager才能在apply_async里使用参数传递Queue对象,
    m=Manager()
    # 创建Queue队列
    q = m.Queue()
    # 创建多个进程，表示可以同时执行的进程数量。默认大小是CPU的核心数
    p = Pool(4)
    New_start = ["http://www.baidu.com","http://www."]
    for news in New_start:
        q.put(news)
    #for i in range(10):
    
    for i in range(5):
        # 创建进程，放入进程池统一管理,并使用异步非阻塞方式运行
        p.apply_async(run, args=(q,))
    # 如果我们用的是进程池，在调用join()之前必须要先close()，并且在close()之后不能再继续往进程池添加新的进程
    p.close()
    # 进程池对象调用join，会等待进程吃中所有的子进程结束完毕再去结束父进程
    p.join()
    
    print("父进程结束。")

[此贴子已经被作者于2019-9-27 11:00编辑过]