0 前言

多线程再平时的树莓派中将会非常广泛——比如控制多个电机同时运动等等 莫凡大神视频地址大神视频教学链接

1 基础

线程开始就是这一步完成，继续下一步，与此同时，分出了一个支线，并且开始支线

import threading import time def job(): print('T1 start\n') for i in range(10): time.sleep(0.1) print('T1 finish\n') #print('number:%s'%threading.current_thread()) def T2(): print('T2 start\n') print('T2 finish\n') def main(): for i in range(4): add_thread = threading.Thread(target=job,name = 'T1') #启动线程1来运行job函数 add_thread2 = threading.Thread(target=T2,name = 'T2') #启动线程2来运行T2函数 add_thread.start() #add_thread启动 add_thread2.start() #add_thread2启动 add_thread.join() #join当add_thread完成后再执行以下操作 print('all done\n') #print(threading.active_count()) #print(threading.enumerate()) #print(threading.current_thread()) main() import threading import time from queue import Queue #queue可以让多线程函数有返回值 def job(l,p): for i in range(len(l)): l[i] = l[i]**2 print('job start\n') print('job finshed\n') p.put(l) #将q放入列表l当中 def mul(): q = Queue() #Queue---放入返回值 thrads = [] data = [[1,2,3],[4,5,6],[7,8,9],[4,5,8]] for i in range(4): #这里用于开启多4个线程分别运算data[i] t = threading.Thread(target = job,args = (data[i],q)) #对已经开启的线程进行传参数（data[i],q） #创建线程 print(t) t.start() #开始第i线程 继续往下走 thrads.append(t) for f in thrads: f.join() print('wancheng???') #等第i个线程结束了继续往下走 results = [] for t in range(4): results.append(q.get()) #q.get()从q中读取一个值（只有按顺序拿出一个） print('append') print(results) mul()

2 控制两台电机

说明：设计了一个class，用来直接调用motor模块进行控制，减小代码量，减小读者阅读难度

import RPi.GPIO as GPIO import time import threading #指定第一个电机的GPIO口 IN1 = 40 # pin40 IN2 = 38 IN3 = 36 IN4 = 35 #指定第二个电机的GPIO口 IN5 = 18 IN6 = 16 IN7 = 13 IN8 = 15 #设置两台电机的频率和步数 motor1_delay = 0.0002 motor1_steps = 4000 motor2_delay = 0.02 motor2_steps = 400 #定义电机的类 class Motor: def __init__(self,I1,I2,I3,I4): #初始化类的参数 self.I1 = I1 self.I2 = I2 self.I3 = I3 self.I4 = I4 GPIO.setwarnings(False) GPIO.setmode(GPIO.BOARD) # 设置寻址模式，按照pin引脚寻址 GPIO.setup(self.I1, GPIO.OUT) # 指定IN1输出 GPIO.setup(self.I2, GPIO.OUT) GPIO.setup(self.I3, GPIO.OUT) GPIO.setup(self.I4, GPIO.OUT) def loop(self,delay_loop,steps_loop): print ("backward...") Motor.backward(self,delay_loop,steps_loop) print ("stop...") Motor.stop(self) # stop time.sleep(3) # sleep 3s print ("forward...") Motor.forward(self,delay_loop,steps_loop) print ("stop...") Motor.stop(self) time.sleep(3) def setStep(self, w1, w2, w3, w4): #GPIO.output（）---指定脚位输出high还是low GPIO.output(self.I1, w1)#.output（in1，1）代表指定IN1引脚输出的是高电平，0就是低电平 GPIO.output(self.I2, w2) GPIO.output(self.I3, w3) GPIO.output(self.I4, w4) def backward(self, delay, steps): for i in range(0, steps): Motor.setStep(self,1, 0, 0, 1) time.sleep(delay) Motor.setStep(self,0, 1, 0, 1) time.sleep(delay) Motor.setStep(self,0, 1, 1, 0) time.sleep(delay) Motor.setStep(self,1, 0, 1, 0) time.sleep(delay) def forward(self,delay, steps): for i in range(0, steps): Motor.setStep(self,1, 0, 1, 0) time.sleep(delay) Motor.setStep(self,0, 1, 1, 0) time.sleep(delay) Motor.setStep(self,0, 1, 0, 1) time.sleep(delay) Motor.setStep(self,1, 0, 0, 1) time.sleep(delay) def stop(self): Motor.setStep(self,0, 0, 0, 0) def destroy(): GPIO.cleanup() if __name__ == '__main__': #主函数测试 motor1 = Motor(IN1,IN2,IN3,IN4)#实例化类 #motor1.setup(IN1,IN2,IN3,IN4) motor2 = Motor(IN5,IN6,IN7,IN8)#实例化类 #motor2.setup(IN5,IN6,IN7,IN8) try: while True: shuru = int(input('input:')) if shuru ==1: thread1 = threading.Thread(target=motor1.loop, args=(motor1_delay,motor1_steps))#创建线程1 thread2 = threading.Thread(target=motor2.loop, args=(motor2_delay,motor2_steps))#创建线程2 thread1.start()#开始线程1 thread2.start()#开始线程2 except KeyboardInterrupt: # When 'Ctrl+C' is pressed, the child function destroy() will be executed. destroy()

3 时间控制

import time import threading def thread1(): start = time.time() for i in range(10): end = time.time() t = end - start if t < 2: print('辅助线程%d\n',i) time.sleep(1) else: break def main(): for j in range(50): add_thread = threading.Thread(target=thread1,name = 'T1') add_thread.start() #print('主线程%d\n',j) time.sleep(2) print(threading.active_count()) main()