python多线程

1 多进程

# 多进程,
import os
import time
from multiprocessing import Process
# 启动时必须在 if __name__ 判断下,windows 必须,其他 无限制
# =================================================
# def func(args):
#     print("子进程:",os.getpid())
#     print("子进程的父进程:",os.getppid())
#     time.sleep(10)
#     print("子进程结束")
# if __name__ =="__main__":
#     p = Process(target=func,args=(1,))  # 注册 并传入元祖 元祖有一个参数要加逗号
#     # p是进程对象
#     p.start() # 开启子进程
#     print("主进程:", os.getpid())
#     print("主进程的父进程:",os.getppid()) # cmd 或者是 pycharm
# 生命周期
#   主进程长:自己执行完结束
#   子进程长:等待子进程结束
# =================================================
# 多进程中的方法
# join
# def fun(arg1,arg2):
#     print('*'*arg1)
#     # time.sleep(5)
#     print('*'*arg2)
# if __name__ == "__main__":
#     p = Process(target=fun,args=(10,20))
#     p.start()
#     # p.join() # 感知子进程结束
#     # time.sleep(1)
#     print("all is stop")
#
#     print("最后的语句")
#     os.walk(r"目录") # 返回 文件夹中文件名字
# =================================================
# def fun():
#     print("xxx")
# if __name__ == "__main__":
#     for i in range(10):
#         p = Process(target=fun)
#         p.start()
#         p.join()  # 停止for循环 进程结束后继续
#         print("for")
#     print("主进程")
# =================================================
# 第二种方法
# class MyProcess(Process):
#     def __init__(self,args):
#         super().__init__()  # 若要传递参数,需要调用父类init
#
#     def run(self):
#         print("子进程",self.__dict__)
#         print(self.pid)
# if __name__ == "__main__":
#     print("主进程:",os.getpid())
#     p1 = MyProcess()
#     p1.start()
# =================================================
# 进程之间数据是隔离,命名空间不通
# def fun():
#     global n
#     n= 0
#     print("pid:%s" %os.getpid(),n)
# if __name__ == "__main__":
#     n=100
#     p = Process(target=fun)
#     p.start()
#     p.join()
#     print(n) # -->100
# =================================================
# 多进程tcp连接
# import socket
# # 客户端
# sk = socket.socket()
# sk.connect(("127.0.0.1",8080))
# sk.send('N好'.encode("utf8"))
# msg = sk.recv(1024).decode("utf8")
# print(msg)
# sk.close()
#
# # 服务端
# def server(conn):
#     ret= "你好".encode("utf8")
#     conn.send(ret)
#     msg = conn.recv(1024).decode("utf8")
#     print(msg)
#     conn.close()
#
# sk = socket.socket()
# sk.bind(("127.0.0.1",8080))
# sk.listen()
# if __name__ == "__main__":
#     while True:
#         conn, addr = sk.accept()
#         p = Process(target=server,args=(conn,))
#         p.start()
# =================================================
# 守护进程
# 默认情况 父进程 等待子进程结束
# p.daemon = True 在start前,设置为守护进程,守护进程随父进程(代码执行完毕)结束
#   若父进程在等待 子进程(非守护进程时) ,若父进程代码完毕,守护进程应该结束
# p.is_alive() 判断进程是否存活
# p.terminate() 终止进程
# =================================================
# 锁
# 未加锁实例:
# 火车票
import json
import time
from multiprocessing import Process
# def show(i):
#     with open('ticket') as f:
#         dic = json.load(f)
#     print('余票: %s'%dic['ticket'])
​
def buy_ticket(i):
    with open('ticket') as f:
        dic = json.load(f)
        time.sleep(0.1)
    if dic['ticket'] > 0 :
        dic['ticket'] -= 1
        print('\033[32m%s买到票了\033[0m'%i)
    else:
        print('\033[31m%s没买到票\033[0m'%i)
    time.sleep(0.1)
    with open('ticket','w') as f:
        json.dump(dic,f)
if __name__ == '__main__':
    # for i in range(10):
    #     p = Process(target=show,args=(i,))
    #     p.start()
    for i in range(10):
        p = Process(target=buy_ticket, args=(i))
        p.start()
# =================================================
# 锁
# 加锁实例
# 火车票
import json
import time
from multiprocessing import Process
from multiprocessing import Lock
​
# def show(i):
#     with open('ticket') as f:
#         dic = json.load(f)
#     print('余票: %s'%dic['ticket'])
​
def buy_ticket(i,lock):
    lock.acquire() #拿钥匙进门
    with open('ticket') as f:
        dic = json.load(f)
        time.sleep(0.1)
    if dic['ticket'] > 0 :
        dic['ticket'] -= 1
        print('\033[32m%s买到票了\033[0m'%i)
    else:
        print('\033[31m%s没买到票\033[0m'%i)
    time.sleep(0.1)
    with open('ticket','w') as f:
        json.dump(dic,f)
    lock.release()      # 还钥匙
​
if __name__ == '__main__':
    # for i in range(10):
    #     p = Process(target=show,args=(i,))
    #     p.start()
    lock = Lock()
    for i in range(10):
        p = Process(target=buy_ticket, args=(i,lock))
        p.start()
# =================================================
# =================================================
# =================================================

2 信号量_事件

# 多进程中的组件
# 一个资源 同一时间 被n个人访问
import time
import random
from multiprocessing import Process,Event
# ==============================
# 未用信号量
# def ktv(i):
#     print('%s走进ktv'%i)
#     time.sleep(random.randint(1,5))
#     print('%s走出ktv'%i)
# if __name__ == '__main__' :
#     for i in range(20):
#         p = Process(target=ktv,args=(i))
#         p.start()
# ==============================
from multiprocessing import Semaphore
​
# sem = Semaphore(4)
# sem.acquire()
# print('拿到第一把钥匙')
# sem.acquire()
# print('拿到第二把钥匙')
# sem.acquire()
# print('拿到第三把钥匙')
# sem.acquire()
# print('拿到第四把钥匙')
# sem.acquire()
# print('拿到第五把钥匙')
# def ktv(i,sem):
#     sem.acquire()    #获取钥匙
#     print('%s走进ktv'%i)
#     time.sleep(random.randint(1,5))
#     print('%s走出ktv'%i)
#     sem.release()    # 释放钥匙
#
#
# if __name__ == '__main__' :
#     sem = Semaphore(4)
#     for i in range(20):
#         p = Process(target=ktv,args=(i,sem))
#         p.start()
​
# ==============================
# 事件
#   信号是控制进程阻塞与否
#   事件创建后,默认是阻塞状态
# e = Event() # 创建事件
# e.is_set()  # False 默认阻塞
# print("xx") # 可打印 e.set() 设置为True e.clear() 设置为False
# e.wait()
# print("xx") # 阻塞
# 遇到wait()会判断is_set() 为False 阻塞
# ==============================
# 红绿灯事件
def cars(e,i):
    if not e.is_set():
        print("car%i在等待" % i)
        e.wait()
    print("car%i通过" % i)
​
​
def light(e):
    while True:
        if e.is_set():
            e.clear()
            print("\033[31m红灯\033[0m")
        else:
            e.set()
            print("\033[32m绿灯\033[0m")
        time.sleep(2)
if __name__ == "__main__":
    e = Event()
    p =Process(target=light,args=(e,))
    p.start()
    for i in range(1,21):
        car = Process(target=cars,args=(e,i))
        car.start()
        time.sleep(random.random())
# ==============================
# ==============================
# ==============================
# ==============================
# def test(e):
#     e.set()
#     print("xxx")
# if __name__=="__main__":
#     e = Event()
#     print(e.is_set())
#     p = Process(target=test,args=(e,))
#     p.start()
#     e.wait()
#     print("注")
​

3 进程通信_队列管道

# IPC 内部进程通信,不能使用普通queue
from multiprocessing import Queue,Process
# ===============================
# q = Queue(5)  # 队列大小
# q.put(1)
# q.put(1)
# q.full()  # 若队列满了,阻塞等待
# q.get()
# q.empty()  # 若为空,阻塞等待有数据 后取值
# q.get()
# q.get_nowait()  # 用于跳过等待,需要用try
# ===============================
# def produce(q):
#     q.put('hello')
# def consume(q):
#     print(q.get())
# if __name__ =="__main__":
#     q = Queue()
#     p = Process(target=produce,args=(q,))
#     p.start()
#     p2 = Process(target=consume, args=(q,))
#     p2.start()
# ===============================
# 生产者消费者模型
# 若生产者,生产有数量,消费者,不停消费,最后消费进程会处于等待状态
# 可在主进程后边join生产进程,消费进程判断为空,但不准确
#   需要在队列put(None) 子进程判断,由于数据之间不能共享,需要put 消费数量的None
# ===============================
from multiprocessing import JoinableQueue
# consume :
#     ....
#     q.task_done()
# produce :
#     ...
#     q.join()
# ===============================
#  循环通知,致使进程结束
# JoinableQueue
# 生产者生产,不停不停消费,若q为空 一直等待,
#   生产者完毕后,会join等待消费值消费完毕,因为是同一个q,一个生产者完毕后,其他还没有完毕q会处于,他会处于阻塞
#   等待 消费者 全部消费完毕,q.join()会感知,因此 生产进程会结束,主进程最后join生产进程,生产结束
#   主进程就结束,身为守护进程的子进程也结束
# import time
# import random
# from multiprocessing import Process,JoinableQueue
# def consumer(q,name):
#     while True:
#         food = q.get()
#         print('\033[31m%s消费了%s\033[0m' % (name,food))
#         time.sleep(random.randint(1,3))
#         q.task_done()     # count - 1
#
# def producer(name,food,q):
#     for i in range(4):
#         time.sleep(random.randint(1,3))
#         f = '%s生产了%s%s'%(name,food,i)
#         print(f)
#         q.put(f)
#     q.join()    # 阻塞  直到一个队列中的所有数据 全部被处理完毕
#
# if __name__  == '__main__':
#     q = JoinableQueue(20)
#     p1 = Process(target=producer,args=('Egon','包子',q))
#     p2 = Process(target=producer, args=('wusir','泔水', q))
#     c1 = Process(target=consumer, args=(q,'alex'))
#     c2 = Process(target=consumer, args=(q,'jinboss'))
#     p1.start()
#     p2.start()
#     c1.daemon = True   # 设置为守护进程 主进程中的代码执行完毕之后,子进程自动结束
#     c2.daemon = True
#     c1.start()
#     c2.start()
#     p1.join()
#     p2.join()      # 感知一个进程的结束
​
#  在消费者这一端:
    # 每次获取一个数据
    # 处理一个数据
    # 发送一个记号 : 标志一个数据被处理成功
​
# 在生产者这一端:
    # 每一次生产一个数据,
    # 且每一次生产的数据都放在队列中
    # 在队列中刻上一个记号
    # 当生产者全部生产完毕之后,
    # join信号 : 已经停止生产数据了
                # 且要等待之前被刻上的记号都被消费完
                # 当数据都被处理完时,join阻塞结束
​
# consumer 中把所有的任务消耗完
# producer 端 的 join感知到,停止阻塞
# 所有的producer进程结束
# 主进程中的p.join结束
# 主进程中代码结束
# 守护进程(消费者的进程)结束
# ===============================
# 管道 双向通信工具
from multiprocessing import Pipe
# conn,conn2 = Pipe()
# conn.send("123456")  # 不用字节
# print(conn2.recv())  # 不用指定大小
​
def fun(conn):
    conn.send("hello")
if __name__=="__main__":
    conn1,conn2  = Pipe()
    Process(target=fun,args=(conn1,)).start()
    print(conn1.recv())
# 管道返回2个连接
#    conn1, conn2
# P    发送 接受
# p2   接受 发送
# 若只发数据,可关闭一端,若取数据的时候,对面已关闭,则报错,看根据此 终止程序
from multiprocessing import Pipe,Process
# def func(conn1,conn2):
#     conn2.close()
#     while True:
#         try :
#             msg = conn1.recv()
#             print(msg)
#         except EOFError:
#             conn1.close()
#             break
#
# if __name__ == '__main__':
#     conn1, conn2 = Pipe()
#     Process(target=func,args = (conn1,conn2)).start()
#     conn1.close()
#     for i in range(20):
#         conn2.send('吃了么')
#     conn2.close()
# ===============================
# from multiprocessing import Lock,Pipe,Process
# def producer(con,pro,name,food):
#     con.close()
#     for i in range(100):
#         f = '%s生产%s%s'%(name,food,i)
#         print(f)
#         pro.send(f)
#     pro.send(None)
#     pro.send(None)
#     pro.send(None)
#     pro.close()
#
# def consumer(con,pro,name,lock):
#     pro.close()
#     while True:
#             lock.acquire()  # 不安全主要是recv的时候,因此两端加锁即可
#             food = con.recv()
#             lock.release()
#             if food is None:
#                 con.close()
#                 break
#             print('%s吃了%s' % (name, food))
# if __name__ == '__main__':
#     con,pro = Pipe()
#     lock= Lock()
#     p = Process(target=producer,args=(con,pro,'egon','泔水'))
#     c1 = Process(target=consumer, args=(con, pro, 'alex',lock))
#     c2 = Process(target=consumer, args=(con, pro, 'bossjin',lock))
#     c3 = Process(target=consumer, args=(con, pro, 'wusir',lock))
#     c1.start()
#     c2.start()
#     c3.start()
#     p.start()
#     con.close()
#     pro.close()
​
# from multiprocessing import Process,Pipe,Lock
#
# def consumer(produce, consume,name,lock):
#     produce.close()
#     while True:
#         lock.acquire()
#         baozi=consume.recv()
#         lock.release()
#         if baozi:
#             print('%s 收到包子:%s' %(name,baozi))
#         else:
#             consume.close()
#             break
#
# def producer(produce, consume,n):
#     consume.close()
#     for i in range(n):
#         produce.send(i)
#     produce.send(None)
#     produce.send(None)
#     produce.close()
#
# if __name__ == '__main__':
#     produce,consume=Pipe()
#     lock = Lock()
#     c1=Process(target=consumer,args=(produce,consume,'c1',lock))
#     c2=Process(target=consumer,args=(produce,consume,'c2',lock))
#     p1=Process(target=producer,args=(produce,consume,30))
#     c1.start()
#     c2.start()
#     p1.start()
#     produce.close()
#     consume.close()
​
# pipe 数据不安全性
# IPC
# 加锁来控制操作管道的行为 来避免进程之间争抢数据造成的数据不安全现象
​
# 队列 进程之间数据安全的
#   管道 + 锁

3_进程通信__数据共享

from multiprocessing import Manager,Process,Lock
# 进程之间不能传递数据,通过方法的参数可以传过去,但修改后,无反应,不知道原因
def main(dict,lock):
    lock.acquire()
    dict['count']-=1
    lock.release()
if __name__ == "__main__":
    m = Manager()
    l = Lock()
    dict = m.dict({"count":100})
    p_list=[]
    for i in range(50):
        p = Process(target=main,args=(dict,l))
        p.start()
        p_list.append(p)
    for i in p_list:i.join()
    print(dict['count'])

4_进程池

# 进程池
# 上一个例子中50个进程很慢
#   寄存器 堆栈 文件
#   操作系统调度,cup切换
# 高级线程池有数量限定 ,有最低,任务变多的时候 逐步加到最高限制
import os,time
from multiprocessing import Pool,Manager
# ==================================
# def func2(i):
#     print(os.getpid(),os.getppid())
#     i+1
# def func(list):
#     list[1]['set'].add(os.getpid())
#     print(len(list[1]['set']))
# # 一般超过5个使用pool
# if __name__ == "__main__":
#     pid = Manager()
#     dict1 = pid.dict({"set":set()})
#     pool = Pool(5)
#     # 执行不同的任务,map 自带join方法
#     #pool.map(func2,range(100))
#     pool.map(func, [[i,dict1]for i in range(100)])
#     print(len(dict1['set']))
# """
# 等于
#     for i in range(100):
#         p = Process(target=func,args=(i,))
#         p.start()
# """
# ==================================
# def fun(n):
#     print("start fun%s" %n,os.getpid())
#     time.sleep(1)
#     print("end fun%s" % n, os.getpid())
# if __name__ == "__main__":
#     p = Pool() # 默认cup核心数量
#     for i in range(10):
#         #p.apply(fun,args=(i,)) # 同步提交的
#         p.apply_async(fun,args=(i,)) # 异步提交,真的异步,因此需要join
#     p.close()  # 不再接受新的任务
#     p.join()  # 感知进程池中任务结束 保持 主进程 与子进程同步
​
# ==================================
# import socket
# from multiprocessing import Pool
#
# def func(conn):
#     conn.send(b'hello')
#     print(conn.recv(1024).decode('utf-8'))
#     conn.close()
#
# if __name__ == '__main__':
#     p = Pool(5)
#     sk = socket.socket()
#     sk.bind(('127.0.0.1',8080))
#     sk.listen()
#     while True:
#         conn, addr = sk.accept()
#         p.apply_async(func,args=(conn,))
#     sk.close()
#     import socket
#
#     sk = socket.socket()
#     sk.connect(('127.0.0.1', 8080))
#
#     ret = sk.recv(1024).decode('utf-8')
#     print(ret)
#     msg = input('>>>').encode('utf-8')
#     sk.send(msg)
#     sk.close()
# ==================================
# 进程池的返回值
# p = Pool()
# p.map(funcname,iterable)     默认异步的执行任务,且自带close和join
# p.apply   同步调用的
# p.apply_async 异步调用 和主进程完全异步 需要手动close 和 join
# from multiprocessing import Pool
# def func(i):
#     return i*i
#
# if __name__ == '__main__':
#     p = Pool(5)
#     for i in range(10):
#         res = p.apply(func,args=(i,))   # apply的结果就是func的返回值
#         print(res) --> 直接就是返回值
# ==================================
# import time
# from multiprocessing import Pool
# def func(i):
#     time.sleep(0.5)
#     return i*i
#
# if __name__ == '__main__':
#     p = Pool(5)
#     res_l = []
#     for i in range(10):
#         res = p.apply_async(func,args=(i,))   # apply的结果就是func的返回值
#         res_l.append(res)
#   若在for 中直接获取res.get()会在成阻塞,程序变同步执行
#     for res in res_l:print(res.get())# 等着 func的计算结果
#    调用res.get时返回
# ==================================
# import time
# from multiprocessing import Pool
# def func(i):
#     time.sleep(0.5)
#     return i*i
#
# if __name__ == '__main__':
#     p = Pool(5)
#     ret = p.map(func,range(100))
#     print(ret)  #  -> 直接返回全部,列表返回
# 自带join,close 最后一起返回
# ====================================
# 回调函数 , 回调的函数在主进程调用
# 对于子进程中再起子进程问题,还不知道
# 每个进程的回调函数 交给主进程顺序执行
import os
from multiprocessing import Pool,Process
def func2(nn):
    print('in func2',os.getpid())
    print(nn)
def func3(n):
    print('in func3', os.getpid())
    return n*n
def func1(n):
    print('in func1',os.getpid())
    p = Pool(5)
    p.apply_async(func3, args=(10,), callback=func2)
    p.close()
    p.join()
    return n*n
if __name__ == '__main__':
    print('主进程 :',os.getpid())
    p = Pool(5)
    p.apply_async(func1,args=(10,),callback=func2)
    p.close()
    p.join()
# ===================================================
import requests
from urllib.request import urlopen
from multiprocessing import Pool
# 200 网页正常的返回
# 404 网页找不到
# 502 504
# 场景:callback 耗时段,远小于网络延时,此时使用,在主进程运行,
def get(url):
    response = requests.get(url)
    if response.status_code == 200:
        return url, response.content.decode('utf-8')
​
​
def get_urllib(url):
    ret = urlopen(url)
    return ret.read().decode('utf-8')
​
​
def call_back(args):
    url, content = args
    print(url, len(content))
​
​
if __name__ == '__main__':
    url_lst = [
        'https://www.cnblogs.com/',
        'http://www.baidu.com',
        'https://www.sogou.com/',
        'http://www.sohu.com/',
    ]
    p = Pool(5)
    for url in url_lst:
        p.apply_async(get, args=(url,), callback=call_back) # callback 中的参数为 get函数的返回值
    p.close()
    p.join()

4_线程

两者之间应该有对应关系1:1 1:n

linux 中的nptl 1对1 线程

# 同一进程的线程间的数据共享的,共享的 共享的
#   可通过直接访问全局变量 global,还需要进程同步
#   创建,切换,撤销 相比进程 消耗小,轻量级
#   进程:资源分配单位,每个进程 至少一个线程
#   线程:cup调度单位
# thread 基本模块,避免使用,可能与threading 冲突
# threading thread的高级版本
# Queue 多线程之间共享数据的数据结构
# 与进程类似,好多方法相同
import time
from threading import Thread
import threading
# def func(n):
#     time.sleep(1)
#     print(n)
# t = Thread(target=func,args=(12,))
# t.daemon = True # 成为"守护线程"
# t.start()
# print("主线程") # 默认情况等待子线程结束
# ===================================
# class MyThread(Thread):
#     def __init__(self,name):
#         super().__init__()
#         self.name = name
#     def run(self):
#         # time.sleep(1)
#         print(self.name)
# MyThread("段志方").start()
# ================================
# GIL 锁的是线程,同一时间 只有一个线程 ,cpython解释器的问题,jpython 就不会
# 对于io密集型 没什么区别,只要io时会切换即可
# 但对于多核cup python 同时只能运行一个cup ,其他语言的会运行多个,因此...
# 即不能通过物理核心数增加速度,不能实现(并行)
# ============================================
# 多线程socket 可以input
# import socket
# from threading import Thread
# def chat(conn):
#     conn.send(b'hello')
#     msg = conn.recv(1024).decode('utf-8')
#     print(msg)
#     conn.close()
# sk = socket.socket()
# sk.bind(('127.0.0.1',8080))
# sk.listen()
# while True:
#     conn,addr = sk.accept()
#     Thread(target=chat,args = (conn,)).start()
# sk.close()
#
# import socket
# sk = socket.socket()
# sk.connect(('127.0.0.1',8080))
# msg = sk.recv(1024)
# print(msg)
# inp = input('>>> ').encode('utf-8')
# sk.send(inp)
# sk.close()
# =========================
# print(threading.current_thread()) # 当前线程
# print(threading.active_count()) # 全部线程,包括主线程
# print(threading.enumerate())  # 列表返回全部线程对象
# ==========================================
# 守护线程
# import time
# from threading import Thread
# def func1():
#     while True:
#         print('*'*10)
#         time.sleep(1)
# def func2():
#     print('in func2')
#     time.sleep(5)
#
# t = Thread(target=func1,)
# t.daemon = True
# t.start()
# t2 = Thread(target=func2,)
# t2.start()
# t2.join()
# print('主线程')
​
# (守护进程)随着(主进程代码)的执行结束而结束
# 守护(线程)会在主线程结束之后等待(其他非守护子线程)的结束才结束
​
# 主进程在执行完自己的代码之后不会立即结束 而是等待子进程结束之后 回收子进程的资源
# import time
# from multiprocessing import Process
# def func():
#     time.sleep(5)
#
# if __name__ == '__main__':
#         Process(target=func).start()
# =========================================
# 线程锁 ,与gil无关
import time
from threading import Lock,Thread
# Lock 互斥锁
# def func(lock):
#     global n
#     lock.acquire()
#     temp = n
#     time.sleep(0.2)
#     n = temp - 1
#     lock.release()
#
# n = 10
# t_lst = []
# lock = Lock()
# for i in range(10):
#     t = Thread(target=func,args=(lock,))
#     t.start()
#     t_lst.append(t)
​
# for t in  t_lst: t.join()
# print(n)
​
# 科学家吃面   还会死锁
​
# noodle_lock  = Lock()
# fork_lock = Lock()
# def eat1(name):
#     noodle_lock.acquire()
#     print('%s拿到面条啦'%name)
#     fork_lock.acquire()
#     print('%s拿到叉子了'%name)
#     print('%s吃面'%name)
#     fork_lock.release()
#     noodle_lock.release()
#
# def eat2(name):
#     fork_lock.acquire()
#     print('%s拿到叉子了'%name)
#     time.sleep(1)
#     noodle_lock.acquire()
#     print('%s拿到面条啦'%name)
#     print('%s吃面'%name)
#     noodle_lock.release()
#     fork_lock.release()
#
# Thread(target=eat1,args=('alex',)).start()
# Thread(target=eat2,args=('Egon',)).start()
# Thread(target=eat1,args=('bossjin',)).start()
# Thread(target=eat2,args=('nezha',)).start()
# ===============================================
from threading import RLock   # 递归锁
fork_lock = noodle_lock  = RLock()
# 一个钥匙串上的两把钥匙,同一个lock 在一个线程中可又多次acquire
# 传给其他线程时 不能被acquire
# def eat1(name):
#     print(name)
#     noodle_lock.acquire()            # 一把钥匙
#     print('%s拿到面条啦'%name)
#     fork_lock.acquire()
#     print('%s拿到叉子了'%name)
#     print('%s吃面'%name)
#     fork_lock.release()
#     noodle_lock.release()
#
# def eat2(name):
#     print(name)
#     fork_lock.acquire()
#     print('%s拿到叉子了'%name)
#     time.sleep(1)
#     noodle_lock.acquire()
#     print('%s拿到面条啦'%name)
#     print('%s吃面'%name)
#     noodle_lock.release()
#     fork_lock.release()
# Thread(target=eat1,args=('alex',)).start()
# Thread(target=eat2,args=('Egon',)).start()
# Thread(target=eat1,args=('bossjin',)).start()
# Thread(target=eat2,args=('nezha',)).start()
# =================================================

5_线程_信号量_事件_条件_定时器_列队_线程池

import time
from threading import Semaphore,Thread
# ====================================
# def func(sem,a,b):
#     sem.acquire()
#     time.sleep(1)
#     print(a+b)
#     sem.release()
# sem = Semaphore(4)
# for i in range(10):
#     t = Thread(target=func,args=(sem,i,i+5))
#     t.start()
# ====================================
# 事件被创建的时候
# False状态
    # wait() 阻塞
# True状态
    # wait() 非阻塞
# clear 设置状态为False
# set  设置状态为True
#  数据库 - 文件夹
#  文件夹里有好多excel表格
    # 1.能够更方便的对数据进行增删改查
    # 2.安全访问的机制
#  起两个线程
#  第一个线程 : 连接数据库
        # 等待一个信号 告诉我我们之间的网络是通的
        # 连接数据库
#  第二个线程 : 检测与数据库之间的网络是否连通
        # time.sleep(0,2) 2
        # 将事件的状态设置为True
# import time
# import random
# from threading import Thread,Event
# def connect_db(e):
#     count = 0
#     while count < 3:
#         e.wait(0.5)   # 状态为False的时候,我只等待1s就结束
#         if e.is_set() == True:
#             print('连接数据库')
#             break
#         else:
#             count += 1
#             print('第%s次连接失败'%count)
#     else:
#         raise TimeoutError('数据库连接超时')
# def check_web(e):
#     time.sleep(random.randint(0,3))
#     e.set()
# e = Event()
# t1 = Thread(target=connect_db,args=(e,))
# t2 = Thread(target=check_web,args=(e,))
# t1.start()
# t2.start()
# ====================================
# 条件 复杂的锁
# 条件
from threading import Condition
# 条件
# 锁
# acquire release
# 一个条件被创建之初 默认有一个(False)状态
# False状态 会影响wait一直处于等待状态
# notify(int数据类型)  造钥匙
# from threading import Thread,Condition
# def func(con,i):
#     con.acquire()
#     con.wait() # 等钥匙
#     print('在第%s个循环里'%i)
#     con.release()
# con = Condition()
# for i in range(10):
#     Thread(target=func,args = (con,i)).start()
# while True:
#     num = int(input('>>>'))
#     con.acquire()
#     con.notify(num)  # 造钥匙
#     con.release()
# ====================================
#定时器
# import time
# from threading import Timer
# def func():
#     print('时间同步')   #1-3
# while True:
#     t = Timer(5,func).start()   # 非阻塞的 ,异步的 ,会把所有的5s在一起
#     time.sleep(5) # 睡5s 每5s进行意思时间同步
# ====================================
# 加锁 麻烦 所以使用队列
#线程通信
# queue
# import queue #直接导入普通queue 是线程安全的
# q = queue.Queue()  # 队列 先进先出
# q.put()
# q.get()
# q.put_nowait()
# q.get_nowait()
# q = queue.LifoQueue()  # 栈 先进后出
# q.put(1)
# q.put(2)
# q.put(3)
# print(q.get())
# print(q.get())
# q = queue.PriorityQueue()  # 优先级队列
# q.put((1,'a'))
# q.put((10,'b'))
# q.put((30,'c'))
# q.put((1,'d'))
# q.put((1,'f'))
# print(q.get())
# 元祖中的元素按顺序比较,数字越小优先级大,祖父按照ascii越小优先级越大
# ====================================
# 线程池
import time
# 以前没有线程池
from concurrent.futures import ThreadPoolExecutor
# ProcessPoolExecutor 该模块下还有一个进程池,与multi 功能相同
# submit(fn,*args,**kwargs) 异步提交任务
# map(fun,*iterables,timeout=None,chunksize - 1) 循环的submit
# shutdown(wait=True) # 等于原来的 close join 合并
# result(time=None) 取得结果
# add_done_callback(fn) 回调函数
def func(n):
    time.sleep(2)
    print(n)
    return n*n
def call_back(m):
    print('结果是 %s'%m.result())
# 若使用进程池 只换ThreadPoolExecutor->ProcessPoolExecutor
tpool = ThreadPoolExecutor(max_workers=5)   #  默认 不要超过cpu个数*5
for i in range(20):
    tpool.submit(func,i).add_done_callback(call_back)
tpool.shutdown()
# tpool.map(func,range(20))  # 拿不到返回值
​
# t_lst = []
# for i in  range(20):
#     t = tpool.submit(func,i)
#     t_lst.append(t)
# tpool.shutdown()  # close+join    #
# print('主线程')
# for t in t_lst:print('***',t.result()) # 拿返回值

6_协程

# 进程 多个进程,操作系统负责
# 线程 不能同一时间多个cup 其他语言可以,但不影响高io
#   开启线程 创建线程 寄存器 堆栈
#   关闭一个线程
# 协程
#   本质是一个线程
#   能够在多个任务间切换,不需要寄存器,堆栈切换
#   任务之间切换时间开销 远小于线程
#   计算任务之间切换消耗也很大,一般都是遇到io的时候切换
#   进程(cup数+1)+线程(cup数*5)+协程(500)  = 50000
#   适合爬虫
# 实现并发的手段
# import time
# 实现在 con,pro之间来回切换
# def consumer():
#     while True:
#         x = yield
#         time.sleep(1)
#         print('处理了数据 :',x)
#
# def producer():
#     c = consumer()
#     next(c)
#     for i in range(10):
#         time.sleep(1)
#         print('生产了数据 :',i)
#         c.send(i)
#
# producer()
# =============================================
# 真正的协程模块就是使用greenlet完成的切换
from greenlet import greenlet
# def eat():
#     print('eating start')
#     g2.switch()
#     print('eating end')
#     g2.switch()
#
# def play():
#     print('playing start')
#     g1.switch()
#     print('playing end')
# g1 = greenlet(eat)  # 必须先有g1 ,g2 函数中才能使用g
# g2 = greenlet(play) # 不会自动切换
# g1.switch()
# ======================================
#  不能感知time.sleep(1)
# 可以感知gevent.sleep(1),在第一行引入 如下from...
# 后边的time 都会经过特殊处理,time.sleep() 就可以被识别
# from gevent import monkey;monkey.patch_all()
# import time
# import gevent
# import threading
# def eat():
#     DummyThread-1 虚拟的线程
#     print(threading.current_thread().getName())
#     print(threading.current_thread())
#     print('eating start')
#     time.sleep(1)
#     print('eating end')
#
# def play():
#     DummyThread-2 虚拟的线程
#     print(threading.current_thread().getName())
#     print(threading.current_thread())
#     print('playing start')
#     time.sleep(1)
#     print('playing end')
#
# g1 = gevent.spawn(eat) # 注册进入,会自动切换,不是操作系统调度
# g2 = gevent.spawn(play) # gevent 负责协程的调度 通过封装的greenlet switch
# g1.join()  gevent 是完全异步的  join等待协程结束
# g2.join()
# 进程和线程的任务切换右操作系统完成
# 协程任务之间的切换由程序(代码)完成,只有遇到协程模块能识别的IO操作,(时间片等不识别)的时候,程序才会进行任务切换,实现并发的效果
# ========================================
# 同步 和 异步
# from gevent import monkey;monkey.patch_all() # 放最前面
# import time
# import gevent
# def task(n):
#     time.sleep(1)
#     print(n)
# def sync(): # 同步
#     for i in range(10):
#         task(i)
# def async(): # 异步
#     g_lst = []
#     for i in range(10):
#         g = gevent.spawn(task,i)
#         g_lst.append(g)
#     gevent.joinall(g_lst)   #两种方法都可
#     for g in g_lst:g.join()
# ======================================
# 协程 : 能够在一个线程中实现并发效果的概念
    #    能够规避一些任务中的IO操作
    #    在任务的执行过程中,检测到IO就切换到其他任务
​
# 多线程 被弱化了
# 协程 在一个线程上 提高CPU 的利用率
# 协程相比于多线程的优势 切换的效率更快
# ==========================================
# 爬虫的例子
# 请求过程中的IO等待
# from gevent import monkey;monkey.patch_all()
# import gevent
# from urllib.request import urlopen    # 内置的模块
# urlopen html时有个格式的 reguests 无格式
# def get_url(url):
#     response = urlopen(url)
#     content = response.read().decode('utf-8')
#     return len(content)
#
# g1 = gevent.spawn(get_url,'http://www.baidu.com')
# g2 = gevent.spawn(get_url,'http://www.sogou.com')
# g3 = gevent.spawn(get_url,'http://www.taobao.com')
# g4 = gevent.spawn(get_url,'http://www.hao123.com')
# g5 = gevent.spawn(get_url,'http://www.cnblogs.com')
# gevent.joinall([g1,g2,g3,g4,g5])
# print(g1.value)
# print(g2.value)
# print(g3.value)
# print(g4.value)
# print(g5.value)
​
# ret = get_url('http://www.baidu.com')
# print(ret)
# ======================================
from gevent import monkey;monkey.patch_all()
import socket
import gevent
def talk(conn):
    conn.send(b'hello')
    print(conn.recv(1024).decode('utf-8'))
    conn.close()
​
sk = socket.socket()
sk.bind(('127.0.0.1',8080))
sk.listen()
while True:
    conn,addr = sk.accept()
    gevent.spawn(talk,conn)
sk.close()
​
import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8080))
print(sk.recv(1024))
msg = input('>>>').encode('utf-8')
sk.send(msg)
sk.close()

7_io模型

阻塞模型

非阻塞模型

io多路复用

# 同步 提交一个任务之后要等待这个任务执行完毕
# 异步 只管提交任务,不等待这个任务执行完毕就可以做其他事情
# 阻塞 recv recvfrom accept
# 非阻塞

# 阻塞   线程   运行状态 --> 阻塞状态 --> 就绪
# 非阻塞

# IO多路复用
    # select机制  Windows  linux  都是操作系统轮询每一个被监听的项,看是否有读操作
    # poll机制    linux          它可以监听的对象比select机制可以监听的数量多
                                 # 随着监听项的增多,导致效率降低
    # epoll机制   linux           更高级,绑定回调函数,
# =================================
# 以前的都是阻塞io
# =================================
# 非阻塞io实例
# import socket
# sk = socket.socket()
# sk.bind(('127.0.0.1',9000))
# sk.setblocking(False)  # 设置不阻塞
# sk.listen()
# conn_l = []
# del_conn = []
# while True:
#     try:
#         conn,addr = sk.accept()  #不阻塞,但是没人连我会报错
#         print('建立连接了:',addr)
#         conn_l.append(conn)
#     except BlockingIOError:
#         for con in conn_l:
#             try:
#                 msg = con.recv(1024)  # 非阻塞,如果没有数据就报错
#                 if msg == b'':   # 若客户端关闭 会发送空消息
#                     del_conn.append(con)
#                     continue
#                 print(msg)
#                 con.send(b'byebye')
#             except BlockingIOError:pass
#         for con in del_conn:
#             con.close()
#             conn_l.remove(con)
#         del_conn.clear()
# # while True : 10000   500  501
#
# import time
# import socket
# import threading
# def func():
#     sk = socket.socket()
#     sk.connect(('127.0.0.1',9000))
#     sk.send(b'hello')
#     time.sleep(1)
#     print(sk.recv(1024))
#     sk.close()
#
# for i in range(2):
#     threading.Thread(target=func).start()
# =================================
# io 多路复用, 监听列表的循环 变为有操作系统执行
import select
import socket

sk = socket.socket()
sk.bind(('127.0.0.1',8000))
sk.setblocking(False)
sk.listen()

read_lst = [sk] # 监听列表
while True:   # [sk,conn]
    # 等待读列表,写列表,修改列表 都必传
    # 返回元祖中3个列表,对应三个list,一般只用第一个
    # r_lst里面就是sk对象
    r_lst,w_lst,x_lst = select.select(read_lst,[],[])
    for i in r_lst:
        if i is sk:
            conn,addr = i.accept()
            read_lst.append(conn)
        else:
            ret = i.recv(1024)
            if ret == b'':
                i.close()
                read_lst.remove(i)
                continue
            print(ret)
            i.send(b'goodbye!')
            import time
            import socket
            import threading


            def func():
                sk = socket.socket()
                sk.connect(('127.0.0.1', 8000))
                sk.send(b'hello')
                time.sleep(3)
                print(sk.recv(1024))
                sk.close()
            for i in range(20):
                threading.Thread(target=func).start()

# =================================
import selectors # 选择合适的多路复用机制
from socket import *

def accept(sk,mask):
    conn,addr=sk.accept()
    sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
    try:
        data=conn.recv(1024)
        if not data:
            print('closing',conn)
            sel.unregister(conn)
            conn.close()
            return
        conn.send(data.upper()+b'_SB')
    except Exception:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()

sk=socket()
sk.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
sk.bind(('127.0.0.1',8088))
sk.listen(5)
sk.setblocking(False) #设置socket的接口为非阻塞

sel = selectors.DefaultSelector()   # 自动选择一个适合我的IO多路复用的机制
sel.register(sk,selectors.EVENT_READ,accept)
#相当于网select的读列表里append了一个sk对象,并且绑定了一个回调函数accept
# 说白了就是 如果有人请求连接sk,就调用accrpt方法

while True:
    events=sel.select() #检测所有的sk,conn,是否有完成wait data阶段
    for sel_obj,mask in events:  # [conn]
        callback=sel_obj.data #callback=read
        callback(sel_obj.fileobj,mask) #read(conn,1)

pymysql

import pymysql
# 连接
conn = pymysql.connect(
    host="106.15.39.74",
    port=3306,
    database="test",
    user="root",
    password="dzf123,.",
    charset="utf8" # 没有"-" 没有
)
cursor = conn.cursor()
sql = "select*from student"
name = "dzf"
password = "123456"
sql = "select * from student where name = %s and password = %s"
ret = cursor.execute(sql,[name,password])
# 自己拼接需要加引号,使用防注入sql不用加引号,参数不能少,多
#print(cursor.lastrowid) # 获取刚插入数据的id 应该就是主键 自增的那个,与名字无关
print(ret) # 返回受影响行数
ret = cursor.fetchall() # 元祖 大元组里边小元祖
print(ret,"a")
ret = cursor.fetchone() # 取一条数据
print(ret,"a")
ret = cursor.fetchone()
print(ret,"a")
# 直接返回一条元素,格式是 小元祖,或只有list中的一个小字典,外边没有元祖或list
# 若连续fetchone() 第一次第一条,第二次第二条,一次向下取
# 若取完后 再次 fetchone() 取不到
# -->(('dzf','1234'),('dzf','1234'))
# 在执行语句前 修改cursor格式
cursor.fetchmany(3) # 在cursor位置接下取3条,大元组中小元祖
# 移动光标
cursor.scroll(1,mode="absolute") # 绝对移动 移到1位置,从2开始 ,
cursor.scroll(1,mode="relative") # 相对移动 原来在3 位置,从4 开始读,现在 移动到4 从5开始读
# 向上移可以使用负的
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) # 指定为字典格式
"""
    [
    {'id':1,'name':'dzf'},
    {'id':1,'name':'dzf'},
    ]
"""

cursor.close()
conn.close()
# ====================================
# 插入数据还是用cursor.execute(),注意提交后conn.commit()
# 若多语句,可能错误,conn.rollback()
# sql2 = "insert into student (name,password) values(%s, %s)"
# ret = cursor.execute(sql2,['123','123'])
# conn.commit()
# 或insert into student (name,password) values(%(name)s, %(pwd)s)
# 下边传入字典excute(sql,{"name":xxx..})
# ====================================
# 批量执行
data = (['12','12'],['23','32'],['32','23']) # 格式必须固定
cursor.executemany(sql,data)  # 内部的for循环
# try 防止异常,要回滚, 会取消以前正确的插入语句
# =================================
# 删除,同理,也要提交
# ================================
# 修改 记得提交

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • python基础

    s="dzf".encode("utf-8") 为bytes类型 (将dzf转为bytes类型,utf-8编码方式)

    Dean0731
  • JAVA 线程

    //Executors 工具类 //ExecutorService 接口 ExecutorService service = Executors.newFix...

    Dean0731
  • ESMAScript6基本介绍

    Dean0731
  • 4.并发编程多线程

    multiprocess模块的完全模仿了threading模块的接口,二者在使用层面,有很大的相似性,因而不再详细介绍

    changxin7
  • Python代码规范之简明概述

    Python现在非常火,语法简单而且功能强大,很多同学都想学Python!所以小的给各位看官们准备了高价值Python学习视频教程及相关电子版书籍,欢迎前来领取...

    python学习教程
  • 跟着实例学习ZooKeeper的用法: 计数器

    这一篇文章我们将学习使用Curator来实现计数器。 顾名思义,计数器是用来计数的, 利用ZooKeeper可以实现一个集群共享的计数器。 只要使用相同的pat...

    用户1263954
  • 一起来学 SpringBoot 2.x | 第七篇:整合 Mybatis

    摘要: 原创出处 http://blog.battcn.com/2018/05/09/springboot/v2-orm-mybatis/ 「唐亚峰」欢迎转载,...

    芋道源码
  • 一起来学SpringBoot | 第七篇:整合Mybatis

    MyBatis 是一款优秀的持久层框架,它支持定制化 SQL、存储过程以及高级映射,几乎避免了所有的 JDBC 代码和手动设置参数以及获取结果集,使用简单的 X...

    battcn
  • Python学习教程(三)

    函数操作 函数是重用的程序段。它们允许你给一块语句一个名称,然后你可以在你的程序的任何地方使用这个名称任意多次地运行这个语句块。这被称为 调用 函数。我们已经使...

    生信宝典
  • Python学习笔记整理(十二)

    一、函数基础 函数可以计算出一个返回值。作用:最大化代码重用,最小化代码冗余,流程的分解 1、函数相关的语句和表达式 语句        例子 Call...

    py3study

扫码关注云+社区

领取腾讯云代金券