多线程与多进程

import time
import threading

def addNum():
    global num #在每一个线程中都获取这个全局变量
    # num-=1
    print("ok")
    lock.acquire()
    temp=num
    time.sleep(0.1)
    num =temp-1  # 对此公共变量进行-1操做
    lock.release()
num = 100  #设定一个共享变量
thread_list = []
lock=threading.Lock()
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)
for t in thread_list: #等待全部线程执行完毕
    t.join()

print('Result: ', num)
#串行

import threading
import time

# mutexA = threading.Lock()
# mutexB = threading.Lock()

Rlock=threading.RLock()

class MyThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):

        self.fun1()
        self.fun2()

    def fun1(self):

        Rlock.acquire()  # 若是锁被占用,则阻塞在这里,等待锁的释放

        print ("I am %s , get res: %s---%s" %(self.name, "ResA",time.time()))

        Rlock.acquire()  # count=2
        print ("I am %s , get res: %s---%s" %(self.name, "ResB",time.time()))
        Rlock.release()   #count-1

        Rlock.release()   #count-1 =0


    def fun2(self):
        Rlock.acquire()  # count=1
        print ("I am %s , get res: %s---%s" %(self.name, "ResB",time.time()))
        time.sleep(0.2)

        Rlock.acquire()  # count=2
        print ("I am %s , get res: %s---%s" %(self.name, "ResA",time.time()))
        Rlock.release()

        Rlock.release()   # count=0


if __name__ == "__main__":

    print("start---------------------------%s"%time.time())

    for i in range(0, 10):

        my_thread = MyThread()
        my_thread.start()

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)

def worker(event):
    logging.debug('Waiting for redis ready...')
    event.wait()
    logging.debug('redis ready, and connect to redis server and do some work [%s]', time.ctime())
    time.sleep(1)

def main():
    readis_ready = threading.Event()
    t1 = threading.Thread(target=worker, args=(readis_ready,), name='t1')
    t1.start()

    t2 = threading.Thread(target=worker, args=(readis_ready,), name='t2')
    t2.start()

    logging.debug('first of all, check redis server, make sure it is OK, and then trigger the redis ready event')
    time.sleep(3) # simulate the check progress
    readis_ready.set()

if __name__=="__main__":
    main()

def worker(event):
    while not event.is_set():
        logging.debug('Waiting for redis ready...')
        event.wait(2)
    logging.debug('redis ready, and connect to redis server and do some work [%s]', time.ctime())
    time.sleep(1)

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)


def worker(event):
    logging.debug('Waiting for redis ready...')

    while not event.isSet():
        logging.debug("wait.......")
        event.wait(3)   # if flag=False阻塞，等待flag=true继续执行


    logging.debug('redis ready, and connect to redis server and do some work [%s]', time.ctime())
    time.sleep(1)

def main():

    readis_ready = threading.Event()  #  flag=False
    t1 = threading.Thread(target=worker, args=(readis_ready,), name='t1')
    t1.start()

    t2 = threading.Thread(target=worker, args=(readis_ready,), name='t2')
    t2.start()

    logging.debug('first of all, check redis server, make sure it is OK, and then trigger the redis ready event')

    time.sleep(6) # simulate the check progress
    readis_ready.set()  # flag=Ture


if __name__=="__main__":
    main()

import threading
import time

semaphore = threading.Semaphore(5)

def func():
    if semaphore.acquire():
        print (threading.currentThread().getName() + ' get semaphore')
        time.sleep(2)
        semaphore.release()

for i in range(20):
  t1 = threading.Thread(target=func)
  t1.start()

import time,random
import queue,threading

q = queue.Queue()

def Producer(name):
  count = 0
  while count <10:
    print("making........")
    time.sleep(random.randrange(3))
    q.put(count)
    print('Producer %s has produced %s baozi..' %(name, count))
    count +=1
    #q.task_done()
    #q.join()
    print("ok......")
def Consumer(name):
  count = 0
  while count <10:
    time.sleep(random.randrange(4))
    if not q.empty():
        data = q.get()
        #q.task_done()
        #q.join()
        print(data)
        print('\033[32;1mConsumer %s has eat %s baozi...\033[0m' %(name, data))
    else:
        print("-----no baozi anymore----")
    count +=1

p1 = threading.Thread(target=Producer, args=('A',))
c1 = threading.Thread(target=Consumer, args=('B',))
# c2 = threading.Thread(target=Consumer, args=('C',))
# c3 = threading.Thread(target=Consumer, args=('D',))
p1.start()
c1.start()
# c2.start()
# c3.start()

# Process类调用

from multiprocessing import Process
import time
def f(name):

    print('hello', name,time.ctime())
    time.sleep(1)

if __name__ == '__main__':
    p_list=[]
    for i in range(3):
        p = Process(target=f, args=('alvin:%s'%i,))
        p_list.append(p)
        p.start()
    for i in p_list:
        p.join()
    print('end')

# 继承Process类调用
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self):
        super(MyProcess, self).__init__()
        # self.name = name

    def run(self):

        print ('hello', self.name,time.ctime())
        time.sleep(1)


if __name__ == '__main__':
    p_list=[]
    for i in range(3):
        p = MyProcess()
        p.start()
        p_list.append(p)

    for p in p_list:
        p.join()

    print('end')

#coding:utf8
from multiprocessing import Process
import time

def counter():
    i = 0
    for _ in range(40000000):
        i = i + 1
    return True
def main():
    l=[]
    start_time = time.time()

    for _ in range(2):
        t=Process(target=counter)
        t.start()
        l.append(t)
        #t.join()

    for t in l:
       t.join()

    # counter()
    # counter()
    end_time = time.time()
    print("Total time: {}".format(end_time - start_time))
if __name__ == '__main__':
    main()

"""
测得时候，注意关闭其余无用的软件。防止出如今多进程环境中串行比并行还快。
这是由于其余进程在干扰。
"""

from multiprocessing import Process
import os
import time
def info(name):


    print("name:",name)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())
    print("------------------")
    time.sleep(1)

def foo(name):

    info(name)

if __name__ == '__main__':

    info('main process line')


    p1 = Process(target=info, args=('alvin',))
    p2 = Process(target=foo, args=('egon',))
    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print("ending")

#输出结果
# name: main process line
# parent process: 5164 #pycharm进程号
# process id: 2584 
# ------------------
# name: alvin
# parent process: 2584
# process id: 8100
# ------------------
# name: egon
# parent process: 2584
# process id: 7752
# ------------------
# ending

import time

"""
传统的生产者-消费者模型是一个线程写消息，一个线程取消息，经过锁机制控制队列和等待，但一不当心就可能死锁。
若是改用协程，生产者生产消息后，直接经过yield跳转到消费者开始执行，待消费者执行完毕后，切换回生产者继续生产，效率极高。
"""
# 注意到consumer函数是一个generator（生成器）:
# 任何包含yield关键字的函数都会自动成为生成器(generator)对象

def consumer():
    r = ''
    while True:
        # 三、consumer经过yield拿到消息，处理，又经过yield把结果传回；
        #    yield指令具备return关键字的做用。而后函数的堆栈会自动冻结(freeze)在这一行。
        #    当函数调用者的下一次利用next()或generator.send()或for-in来再次调用该函数时，
        #    就会从yield代码的下一行开始，继续执行，再返回下一次迭代结果。经过这种方式，迭代器能够实现无限序列和惰性求值。
        n = yield r
        if not n:
            return
        print('[CONSUMER] ←← Consuming %s...' % n)
        time.sleep(1)
        r = '200 OK'
def produce(c):
    # 一、首先调用c.next()启动生成器
    next(c)
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] →→ Producing %s...' % n)
        # 二、而后，一旦生产了东西，经过c.send(n)切换到consumer执行；
        cr = c.send(n)
        # 四、produce拿到consumer处理的结果，继续生产下一条消息；
        print('[PRODUCER] Consumer return: %s' % cr)
    # 五、produce决定不生产了，经过c.close()关闭consumer，整个过程结束。
    c.close()
if __name__=='__main__':
    # 六、整个流程无锁，由一个线程执行，produce和consumer协做完成任务，因此称为“协程”，而非线程的抢占式多任务。
    c = consumer()
    produce(c)
    
    
'''
result:

[PRODUCER] →→ Producing 1...
[CONSUMER] ←← Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 2...
[CONSUMER] ←← Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 3...
[CONSUMER] ←← Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 4...
[CONSUMER] ←← Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 5...
[CONSUMER] ←← Consuming 5...
[PRODUCER] Consumer return: 200 OK
'''

from greenlet import greenlet
 
def test1():
    print (12)
    gr2.switch()
    print (34)
    gr2.switch()
 
def test2():
    print (56)
    gr1.switch()
    print (78)
 
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

import gevent
import time

def foo():
    print("running in foo")
    gevent.sleep(2)
    print("switch to foo again")

def bar():
    print("switch to bar")
    gevent.sleep(5)
    print("switch to bar again")

start=time.time()

gevent.joinall(
    [gevent.spawn(foo),
    gevent.spawn(bar)]
)

print(time.time()-start)

from gevent import monkey
monkey.patch_all()
import gevent
from urllib import request
import time

def f(url):
    print('GET: %s' % url)
    resp = request.urlopen(url)
    data = resp.read()
    print('%d bytes received from %s.' % (len(data), url))

start=time.time()

gevent.joinall([
        gevent.spawn(f, 'https://itk.org/'),
        gevent.spawn(f, 'https://www.github.com/'),
        gevent.spawn(f, 'https://zhihu.com/'),
])

# f('https://itk.org/')
# f('https://www.github.com/')
# f('https://zhihu.com/')

print(time.time()-start)

1、关于Linux的epoll机制：

epoll是Linux内核为处理大批量文件描述符而做了改进的poll，是Linux下多路复用IO接口select/poll的
加强版本，它能显著提升程序在大量并发链接中只有少许活跃的状况下的系统CPU利用率。epoll的优势：

（1）支持一个进程打开大数目的socket描述符。select的一个进程所打开的FD由FD_SETSIZE的设置来限定，而epoll没有这个限制，它所支持的FD上限是
最大可打开文件的数目，远大于2048。

（2）IO效率不随FD数目增长而线性降低：因为epoll只会对“活跃”的socket进行操做，因而，只有”活跃”的socket才会主动去调用 callback函数，其余
idle状态的socket则不会。

（3）使用mmap加速内核与用户空间的消息传递。epoll是经过内核于用户空间mmap同一块内存实现的。

（4）内核微调。

2、libev机制

提供了指定文件描述符事件发生时调用回调函数的机制。libev是一个事件循环器：向libev注册感兴趣的事件，好比socket可读事件，libev会对所注册的事件
的源进行管理，并在事件发生时触发相应的程序。

ps

import time
import socket
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sk.setsockopt
sk.bind(('127.0.0.1',6667))
sk.listen(5)
sk.setblocking(False)
while True:
    try:
        print ('waiting client connection .......')
        connection,address = sk.accept()   # 进程主动轮询
        print("+++",address)
        client_messge = connection.recv(1024)
        print(str(client_messge,'utf8'))
        connection.close()
    except Exception as e:
        print (e)
        time.sleep(4)

#############################client

import time
import socket
sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)

while True:
    sk.connect(('127.0.0.1',6667))
    print("hello")
    sk.sendall(bytes("hello","utf8"))
    time.sleep(2)
    break

import socket
import select

sock = socket.socket()
sock.bind(("127.0.0.1",8800))
sock.listen(5)

sock.setblocking(False)
inputs=[sock,]
while 1:
    r,w,e=select.select(inputs,[],[]) # 监听有变化的套接字 inputs=[sock,conn1,conn2,conn3..]
    #r=inputs  r=[conn1,conn2]
    print(inputs,"===inputs===") #必定要注意，r不等于inputs,r是会变化得
    print(r,"====r===")
    for obj in r: # 第一次 [sock,]  第二次 #[conn1,]
        if obj==sock:
            conn,addr=obj.accept()
            print(conn,"===conn===")
            inputs.append(conn) #  inputs=[sock,conn]
        else:
            data=obj.recv(1024)
            print(data.decode("utf8"))
            send_data = input(">>>")
            obj.send(send_data.encode("utf8"))

#输出结果
# [<socket.socket fd=204, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8800)>] ===inputs===
# [<socket.socket fd=204, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8800)>] ====r===
# <socket.socket fd=196, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8800), raddr=('127.0.0.1', 61457)> ===conn===
# [<socket.socket fd=204, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8800)>, <socket.socket fd=196, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8800), raddr=('127.0.0.1', 61457)>] ===inputs===
# [<socket.socket fd=196, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 8800), raddr=('127.0.0.1', 61457)>] ====r===
# aaa #接收得数据
# >>>bbb #客户端发送数据

import socket

sock=socket.socket()

sock.connect(("127.0.0.1",8800))

while 1:
    data=input("input>>>")
    sock.send(data.encode("utf8"))
    rece_data=sock.recv(1024)
    print(rece_data.decode("utf8"))
sock.close()

#输入结果
#input>>>aaa
#bbb
#input>>>