澳门游戏平台大全 澳门十大电子游戏平台 澳门十大电子游戏平台何以在Python中编辑并发程序,爬虫分类总括

澳门十大电子游戏平台何以在Python中编辑并发程序,爬虫分类总括



如何在Python中编写并发程序,python编写并发

GIL

在Python中,由于历史原因(GIL),使得Python中多线程的效果非常不理想.GIL使得任何时刻Python只能利用一个CPU核,并且它的调度算法简单粗暴:多线程中,让每个线程运行一段时间t,然后强行挂起该线程,继而去运行其他线程,如此周而复始,直到所有线程结束.

这使得无法有效利用计算机系统中的”局部性”,频繁的线程切换也对缓存不是很友好,造成资源的浪费.

据说Python官方曾经实现了一个去除GIL的Python解释器,但是其效果还不如有GIL的解释器,遂放弃.后来Python官方推出了”利用多进程替代多线程”的方案,在Python3中也有concurrent.futures这样的包,让我们的程序编写可以做到”简单和性能兼得”.

多进程/多线程+Queue

一般来说,在Python中编写并发程序的经验是:计算密集型任务使用多进程,IO密集型任务使用多进程或者多线程.另外,因为涉及到资源共享,所以需要同步锁等一系列麻烦的步骤,代码编写不直观.另外一种好的思路是利用多进程/多线程+Queue的方法,可以避免加锁这样麻烦低效的方式.

现在在Python2中利用Queue+多进程的方法来处理一个IO密集型任务.

假设现在需要下载多个网页内容并进行解析,单进程的方式效率很低,所以使用多进程/多线程势在必行.
我们可以先初始化一个tasks队列,里面将要存储的是一系列dest_url,同时开启4个进程向tasks中取任务然后执行,处理结果存储在一个results队列中,最后对results中的结果进行解析.最后关闭两个队列.

下面是一些主要的逻辑代码.

# -*- coding:utf-8 -*-

#IO密集型任务
#多个进程同时下载多个网页
#利用Queue+多进程
#由于是IO密集型,所以同样可以利用threading模块

import multiprocessing

def main():
  tasks = multiprocessing.JoinableQueue()
  results = multiprocessing.Queue()
  cpu_count = multiprocessing.cpu_count() #进程数目==CPU核数目

  create_process(tasks, results, cpu_count)  #主进程马上创建一系列进程,但是由于阻塞队列tasks开始为空,副进程全部被阻塞
  add_tasks(tasks) #开始往tasks中添加任务
  parse(tasks, results) #最后主进程等待其他线程处理完成结果


def create_process(tasks, results, cpu_count):
  for _ in range(cpu_count):
    p = multiprocessing.Process(target=_worker, args=(tasks, results)) #根据_worker创建对应的进程
    p.daemon = True #让所有进程可以随主进程结束而结束
    p.start() #启动

def _worker(tasks, results):
  while True:  #因为前面所有线程都设置了daemon=True,故不会无限循环
    try:
      task = tasks.get()  #如果tasks中没有任务,则阻塞
      result = _download(task)
      results.put(result)  #some exceptions do not handled
    finally:
      tasks.task_done()

def add_tasks(tasks):
  for url in get_urls(): #get_urls() return a urls_list
    tasks.put(url)

def parse(tasks, results):
  try: 
    tasks.join()
  except KeyboardInterrupt as err:
    print "Tasks has been stopped!"
    print err

  while not results.empty():
    _parse(results)



if __name__ == '__main__':
  main()

利用Python3中的concurrent.futures包

在Python3中可以利用concurrent.futures包,编写更加简单易用的多线程/多进程代码.其使用感觉和Java的concurrent框架很相似(借鉴?)
比如下面的简单代码示例

def handler():
  futures = set()

  with concurrent.futures.ProcessPoolExecutor(max_workers=cpu_count) as executor:
    for task in get_task(tasks):
      future = executor.submit(task)
      futures.add(future)

def wait_for(futures):
  try:
    for future in concurrent.futures.as_completed(futures):
      err = futures.exception()
      if not err:
        result = future.result()
      else:
        raise err
  except KeyboardInterrupt as e:
    for future in futures:
      future.cancel()
    print "Task has been canceled!"
    print e
  return result

总结

要是一些大型Python项目也这般编写,那么效率也太低了.在Python中有许多已有的框架使用,使用它们起来更加高效.
但是自己的一些”小打小闹”的程序这样来编写还是不错的.:)

Python多线程基础模块

1.thread

# -*- coding: UTF-8 -*-
import thread
import time

# 为线程定义一个函数
def print_time( threadName, delay):
   count = 0
   while count < 5:
      time.sleep(delay)
      count += 1
      print "%s: %s" % ( threadName, time.ctime(time.time()) )

# 创建两个线程
try:
   thread.start_new_thread( print_time, ("Thread-1", 2, ) )
   thread.start_new_thread( print_time, ("Thread-2", 4, ) )
except:
   print "Error: unable to start thread"

while 1:
   pass

print "Main Finished"

结果:

Thread-1: Thu Jan 22 16:42:17 2018
Thread-1: Thu Jan 22 16:42:19 2018
Thread-2: Thu Jan 22 16:42:19 2018
Thread-1: Thu Jan 22 16:42:21 2018
Thread-2: Thu Jan 22 16:42:23 2018
Thread-1: Thu Jan 22 16:42:23 2018
Thread-1: Thu Jan 22 16:42:25 2018
Thread-2: Thu Jan 22 16:42:27 2018
Thread-2: Thu Jan 22 16:42:31 2018
Thread-2: Thu Jan 22 16:42:35 2018

while 1: 
    pass

表示死循环,让主线程一直等待。此时”Main Finished”不会输出。

thread.start_new_thread(线程函数,线程函数的参数,tuple型,可选参数)
start_new_thread()要求一定要有前两个参数。所以,就算我们想要运行的函数不要参数,我们也要传一个空的元组。

2.threading

# -*- coding: UTF-8 -*-
import threading

def func(a):
    print "input is :" + a


threads = []
for i in range(10):
    threads.append(threading.Thread(target=func, args=str(i)))

for i in threads:
    i.start()

结果:

input is :0
input is :1
input is :2
input is :3
input is :4
input is :5
input is :6
input is :7
input is :8
input is :9

低级别的thread模块是推荐给高手用,一般应用程序推荐使用更高级的threading模块

1.它更先进,有完善的线程管理支持,此外,在thread模块的一些属性会和threading模块的这些属性冲突。
2.thread模块有很少的(实际上是一个)同步原语,而threading却有很多。
3.thread模块没有很好的控制,特别当你的进程退出时,
比如:当主线程执行完退出时,其他的线程都会无警告,无保存的死亡,
而threading会允许默认,重要的子线程完成后再退出,它可以特别指定daemon类型的线程。

常用实现方式:

#-*- encoding: gb2312 -*-
import string, threading, time

def thread_main(a):
    global count, mutex
    # 获得线程名
    threadname = threading.currentThread().getName()

    for x in xrange(0, int(a)):
        # 取得锁
        mutex.acquire()
        count = count + 1
        # 释放锁
        mutex.release()
        print threadname, x, count
        time.sleep(1)

def main(num):
    global count, mutex
    threads = []

    count = 1
    # 创建一个锁
    mutex = threading.Lock()
    # 先创建线程对象
    for x in xrange(0, num):
        threads.append(threading.Thread(target=thread_main, args=(10,)))
    # 启动所有线程
    for t in threads:
        t.start()
    # 主线程中等待所有子线程退出
    for t in threads:
        t.join()  


if __name__ == '__main__':
    num = 4
    # 创建4个线程
    main(4)

#-*- encoding: gb2312 -*-
import threading
import time

class Test(threading.Thread):
    def __init__(self, num):
        threading.Thread.__init__(self)
        self._run_num = num

    def run(self):
        global count, mutex
        threadname = threading.currentThread().getName()

        for x in xrange(0, int(self._run_num)):
            mutex.acquire()
            count = count + 1
            mutex.release()
            print threadname, x, count
            time.sleep(1)

if __name__ == '__main__':
    global count, mutex
    threads = []
    num = 4
    count = 1
    # 创建锁
    mutex = threading.Lock()
    # 创建线程对象
    for x in xrange(0, num):
        threads.append(Test(10))
    # 启动线程
    for t in threads:
        t.start()
    # 等待子线程结束
    for t in threads:
        t.join()  

您可能感兴趣的文章:

  • python thread 并发且顺序运行示例
  • Python中的并发编程实例
  • python监控网站运行异常并发送邮件的方法
  • 使用Python中的greenlet包实现并发编程的入门教程
  • Python多进程并发(multiprocessing)用法实例详解

GIL
在Python中,由于历史原因(GIL),使得Python中多线程的效果非常不理想.GIL使得任何时刻Python只能利…

基础

python多线程是真正的线程,但却是鸡肋。

有一个GIL锁:Global Interpreter
Lock(全局解释器锁),任何Python线程执行前,必须先获得GIL锁(“通行证”),然后,每执行100条字节码,解释器就自动释放GIL锁,执行别的线程。这个GIL全局锁实际上把所有线程的执行代码都给上了锁,所以,多线程在Python中只能交替执行,即使100个线程跑在100核CPU上,也只能用到1个核。

多线程并没有大大的缩短执行的时间,缩短的效果不是很明显,在单核CPU下的多线程其实都只是并发,不是并行,并发和并行从宏观上来讲都是同时处理多路请求的概念。但并发和并行又有区别,并行是指两个或者多个事件在同一时刻发生;而并发是指两个或多个事件在同一时间间隔内发生

但是python多线程并非完全没用。
对于CPU密集型代码(循环、计数等),计算多,ticks计数很快就会达到阈值,触发GIL的释放与再竞争,切换线程消耗资源。但是对于IO密集型代码(文件处理、网络爬虫等)则有提升效果。


不知道你的代码到底算CPU密集型还是IO密集型???

IO密集型任务执行期间,99%的时间都花在IO上,花在CPU上的时间很少,因此,用运行速度极快的C语言替换用Python这样运行速度极低的脚本语言,完全无法提升运行效率。对于IO密集型任务,最合适的语言就是开发效率最高(代码量最少)的语言,脚本语言是首选,C语言最差。

multiprocessing这个module有一个dummy的sub
module,它是基于multithread实现了multiprocessing的API。

假设你使用的是multiprocessing的Pool,是使用多进程实现concurrency。

from multiprocessing import Pool

如果把这个代码改成下面这样,就变成多线程实现concurrency。

from multiprocessing.dummy import Pool

两种方式都跑一下,哪个速度快用哪个就行了。


线程同步

如果多个线程共同对某个数据修改,则可能出现不可预料的结果,为了保证数据的正确性,需要对多个线程进行同步。使用Thread对象的Lock和Rlock可以实现简单的线程同步,这两个对象都有acquire方法和release方法,对于那些需要每次只允许一个线程操作的数据,可以将其操作放到acquire和release方法之间

import threading
import time

class myThread (threading.Thread):
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter
    def run(self):
        print "Starting " + self.name
       # 获得锁,成功获得锁定后返回True
       # 可选的timeout参数不填时将一直阻塞直到获得锁定
       # 否则超时后将返回False
        threadLock.acquire()
        print_time(self.name, self.counter, 3)
        # 释放锁
        threadLock.release()

def print_time(threadName, delay, counter):
    while counter:
        time.sleep(delay)
        print "%s: %s" % (threadName, time.ctime(time.time()))
        counter -= 1

threadLock = threading.Lock()
threads = []

# 创建新线程
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# 开启新线程
thread1.start()
thread2.start()

# 添加线程到线程列表
threads.append(thread1)
threads.append(thread2)

# 等待所有线程完成
for t in threads:
    t.join()

print "Exiting Main Thread"

在上面的代码中运用了线程锁还有join等待。

运行结果:

Starting Thread-1
Starting Thread-2
Thread-1: Thu Nov  3 18:56:49 2016
Thread-1: Thu Nov  3 18:56:50 2016
Thread-1: Thu Nov  3 18:56:51 2016
Thread-2: Thu Nov  3 18:56:53 2016
Thread-2: Thu Nov  3 18:56:55 2016
Thread-2: Thu Nov  3 18:56:57 2016
Exiting Main Thread

下面是未做线程同步处理的代码:

# -*- coding: UTF-8 -*-

import threading
import time

import thread

exitFlag = 0

class myThread (threading.Thread):   #继承父类threading.Thread
    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter
    def run(self):                   #把要执行的代码写到run函数里面 线程在创建后会直接运行run函数
        print "Starting " + self.name
        print_time(self.name, self.counter, 5)
        print "Exiting " + self.name

def print_time(threadName, delay, counter):
    while counter:
        if exitFlag:
            thread.exit()
        time.sleep(delay)
        print "%s: %s" % (threadName, time.ctime(time.time()))
        counter -= 1

# 创建新线程
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# 开启线程
thread1.start()
thread2.start()

print "Exiting Main Thread"

结果:

Starting Thread-1Starting Thread-2

Exiting Main Thread
Thread-1: Thu Nov  3 18:42:19 2016
Thread-2: Thu Nov  3 18:42:20 2016
Thread-1: Thu Nov  3 18:42:20 2016
Thread-1: Thu Nov  3 18:42:21 2016
Thread-2: Thu Nov  3 18:42:22 2016
Thread-1: Thu Nov  3 18:42:22 2016
Thread-1: Thu Nov  3 18:42:23 2016
Exiting Thread-1
Thread-2: Thu Nov  3 18:42:24 2016
Thread-2: Thu Nov  3 18:42:26 2016
Thread-2: Thu Nov  3 18:42:28 2016
Exiting Thread-2

输出比较混乱。

澳门十大电子游戏平台 1

线程优先级队列

Python的Queue模块中提供了同步的、线程安全的队列类,包括FIFO(先入先出)队列Queue,LIFO(后入先出)队列LifoQueue,和优先级队列PriorityQueue。这些队列都实现了锁原语,能够在多线程中直接使用。可以使用队列来实现线程间的同步。

Queue模块中的常用方法:

Queue.qsize()       返回队列的大小
Queue.empty()       如果队列为空,返回True,反之False
Queue.full()        如果队列满了,返回True,反之False
                    Queue.full 与 maxsize 大小对应

Queue.get([block[, timeout]])   获取队列,timeout等待时间
Queue.get_nowait()              相当Queue.get(False)
Queue.put(item)                 写入队列,timeout等待时间
Queue.put_nowait(item)          相当Queue.put(item, False)
Queue.task_done()               在完成一项工作之后,Queue.task_done()函数向任务已经完成的队列发送一个信号
Queue.join()                    实际上意味着等到队列为空,再执行别的操作

FIFO队列的例子

# -*- coding: UTF-8 -*-

import Queue
import threading
import time

exitFlag = 0

class myThread (threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q
    def run(self):
        print "Starting " + self.name
        process_data(self.name, self.q)
        print "Exiting " + self.name

def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print "%s processing %s" % (threadName, data)
        else:
            queueLock.release()
        time.sleep(1)

threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = Queue.Queue(10)
threads = []
threadID = 1

# 创建新线程
for tName in threadList:
    thread = myThread(threadID, tName, workQueue)
    thread.start()
    threads.append(thread)
    threadID += 1

# 填充队列
queueLock.acquire()
for word in nameList:
    workQueue.put(word)
queueLock.release()

# 等待队列清空
while not workQueue.empty():
    pass

# 通知线程是时候退出
exitFlag = 1

# 等待所有线程完成
for t in threads:
    t.join()
print "Exiting Main Thread"
标签:,

相关文章

发表评论

电子邮件地址不会被公开。 必填项已用*标注

网站地图xml地图