开启MicroPython多线程模式
MicroPython官方版本对多线程的支持
修改后的MicroPython多线程示例
代码运行效果

MicroPython官方版本对多线程的支持
MicroPython官方版本尝试对 多线程 的支持，但是目前的支持只停留在非常初级的阶段，离真正可用还是有一段距离。在尝试增加多线程的支持过程中踩坑无数，不过最后总算成功实现所需要的功能。
修改后的MicroPython多线程示例
多线程的基础上增加了线程锁和信号量的支持：
from umachine import Pin
import utime
import _thread

lock = _thread.allocate_lock()
semp = _thread.allocate_semephare()

定义4个用户线程，分别是信号量演示线程/i2c演示线程/LED演示线程和MQTT消息发布线程：
def sem_thread():
	while True:
        semp.pend()
        print('sem_thread run')

def i2c_thread():
	while True:
        print('i2c_thread read 0x57')
        i2c0.readfrom_mem(0x57, 0, 4)
        utime.sleep(10)
        print('i2c_thread read 0x5f')
        i2c0.readfrom_mem(0x5f, 0x9a, 6)
        utime.sleep(10)
        #_thread.exit()

def led_thread(time_):
	while True:
        print('led_thread on')
        led0.value(1)
        utime.sleep(time_)
        print('led_thread off')
        led0.value(0)
        utime.sleep(time_)
        semp.post()

def mqtt_thread(time_):
	while True:
        lock.acquire()
        print('mqtt_thread message 1')
        mqtt.publish('/home/bedroom/lamp', 'led on')
        mqtt.publish('/home/bedroom/speaker', 'music off')
        utime.sleep(time_)
        print('mqtt_thread message 2')
        mqtt.publish('/home/bedroom/lamp', 'led off')
        mqtt.publish('/home/bedroom/speaker', 'play music')
        utime.sleep(1)
        print('mqtt_thread message 3')
        mqtt.publish('/smart_home/bedroom/window', 'close window')
        utime.sleep(time_)
        lock.release()
        #_thread.exit()

MQTT消息订阅回调函数和连接成功回调函数：
def callback_on_connect(userdata, flags, rc):
        mqtt.subscribe('/home/bedroom/msgbox', 0)

def callback_on_message(userdata, message):
        print(message)
        mqtt.publish('/home/bedroom/air', 'air turn on')

Wi-Fi连接到AP的Python代码(自己写的ATWINC1500 MicroPython库)：
from winc1500 import WLAN
wlan = WLAN(STA_IF)
wlan.connect('KSLINxxxxxx', 'xxxxxxxxx', AUTH_WPA_PSK)

MQTT的连接和订阅(参考前面回调函数)：
from winc1500 import MQTT
mqtt = MQTT(MQTT_CLIENT)
mqtt.username_pw_set('winc_mp_mqtt', '')
mqtt.on_connect(callback_on_connect)
mqtt.on_message(callback_on_message)
mqtt.connect('iot.eclipse.org', 1883, 30)

最后是开始启动线程的操作
_thread.start_new_thread(led_thread, (2,))
_thread.start_new_thread(i2c_thread, ())
_thread.start_new_thread(sem_thread, ())
_thread.start_new_thread(mqtt_thread,(3,))

while True:
        pass

代码运行效果
补充：MicroPython代码运行在Microchip SAMV71-XULT+ATWINC1500(Wi-Fi模组)

多线程设计模式之保护性暂停模式
定义
保护性暂停模式（Guarded Suspension Design Pattern）：当线程在访问某个对象时，发现条件不满足时，就暂时挂起等待条件满足时再次访问。
如果某个结果需要在多线程之间传递，则可以让这些线程关联到一个对象GuardedObject，但是如果这个结果需要不断的从一个线程到另一个线程那么可以使用消息队列（生产者/消费者）
Thread.join()、Future就采用了保护性暂停模式。
简单实现
package com.morris.concurrent.pattern.guardedsuspension.v2;

public class GuardedObject<V> {

    private final Object lock = new Object();

    private V v;

    private boolean isFinished;

    public void set(V v) {
        synchronized (lock) {
            this.v = v;
            isFinished = true;
            lock.notifyAll();
        }
    }

    public V get() throws InterruptedException {
        synchronized (lock) {
            while (!isFinished) {
                lock.wait();
            }
        }
        return v;
    }
}

超时实现
public V get(long millis) {
    synchronized (lock) {
        while (!isFinished) {
            try {
                lock.wait(millis);
                break;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    return v;
}

上面的超时实现有点小问题。
分析：假如超时时间是5s，但是在第2s的时候别人把这个线程中断了但是此时还没有结果，那么这个线程就会进入下次循环继续等待5s，这样这个线程就等待7s了。
最终实现如下：
public V get(long millis) {
    long begin = System.currentTimeMillis(); // 等待开始时间
    long waitedTime = 0; // 等待了多长时间
    synchronized (lock) {
        while (!isFinished) {
            if(waitedTime >= millis) {
                System.out.println("超时了，不等了");
                break;
            }
            try {
                lock.wait(millis - waitedTime); // 继续等待剩余时间
                break;
            } catch (InterruptedException e) {
                waitedTime = System.currentTimeMillis() - begin;
                System.out.println("被中断了，等了" + waitedTime + "ms");
            }
        }
    }
    return v;
}

使用保护性暂停模式实现Future和Callable
public interface Callable<T> {
    T call();
}

public interface Future<T> {
    T get();
}

public class FutureTask<T> implements Future<T> {

    private boolean isFinish;

    private T t;

    public synchronized void set(T t) {
        this.t = t;
        this.isFinish = true;
        this.notifyAll();
    }

    @Override
    public synchronized T get() {
        while (!isFinish) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return t;
    }
}

public abstract class Executors {

    public static <T> Future<T> newSingleThreadPool(Callable<T> callable) {
        FutureTask<T> future = new FutureTask<>();
        new Thread(()->{
            T t = callable.call();
            future.set(t);
        }).start();
        return future;
    }

}

单例模式多线程下不安全

大厂面试题：

1、请你谈谈对volatile的理解？

2、CAS你知道吗？

3、原子类AtomicInteger的ABA问题谈谈？原子更新引用知道吗？

4、我们都知道ArrayList是线程不安全的，请编码写一个不安全的案例并给出解决方案？

5、公平锁/非公平锁/可重入锁/递归锁/自旋锁谈谈你的理解？请手写一个自旋锁。

6、CountDownLatch、CyclicBarrier、Semaphore使用过吗？

7、阻塞队列知道吗？

8、线程池用过吗？ThreadPoolExecutor谈谈你的理解？

9、线程池用过吗？生产上你是如何设置合理参数？

10、死锁编码及定位分析？

 

1、单例模式单线程下是安全的

public class SingletonDemo {

    private static SingletonDemo singletonDemo = null;

 

    private SingletonDemo() {

        System.out.println(Thread.currentThread().getName() + "\t" + "SingletonDemo构造方法创建");

    }

 

    public static SingletonDemo getInstance() {

        if (singletonDemo == null) {

            singletonDemo = new SingletonDemo();

        }

        return singletonDemo;

    }

 

    public static void main(String[] args) {

// 单线程main情况下，单例模式下只会创建一次SingletonDemo对象

System.out.println(SingletonDemo.getInstance() == SingletonDemo.getInstance());

System.out.println(SingletonDemo.getInstance() == SingletonDemo.getInstance());

System.out.println(SingletonDemo.getInstance() == SingletonDemo.getInstance());

    }

}

程序执行结果如下：



 

2、单例模式多线程是不安全的

public class SingletonDemo {

    private static SingletonDemo singletonDemo = null;

 

    private SingletonDemo() {

        System.out.println(Thread.currentThread().getName() + "\t" + "SingletonDemo构造方法创建");

    }

 

    public static SingletonDemo getInstance() {

        if (singletonDemo == null) {

            singletonDemo = new SingletonDemo();

        }

        return singletonDemo;

    }

 

    public static void main(String[] args) {

        // 多线程情况下，单例模式存在线程不安全，会创建多次SingletonDemo对象

        for (int i = 0; i < 10; i++) {

            new Thread(() -> {

                SingletonDemo.getInstance();

            }, String.valueOf(i)).start();

        }

 

    }

}

程序执行结果如下： 



3、单例模式多线程不安全问题解决

public class SingletonDemo {

    private static SingletonDemo singletonDemo = null;

 

    private SingletonDemo() {

        System.out.println(Thread.currentThread().getName() + "\t" + "SingletonDemo构造方法创建");

    }

 

    public static synchronized SingletonDemo getInstance() {

        if (singletonDemo == null) {

            singletonDemo = new SingletonDemo();

        }

        return singletonDemo;

    }

 

    public static void main(String[] args) {

        // 多线程情况下，单例模式存在线程不安全，会创建多次SingletonDemo对象

        for (int i = 0; i < 10; i++) {

            new Thread(() -> {

                SingletonDemo.getInstance();

            }, String.valueOf(i)).start();

        }

 

    }

}

程序执行结果如下：



volatile使用场景请接着看下一篇博客

参考答案请看下一小节：单例模式volatile分析

 

一.本例来自《java多线程编程调试模式》：
题意：模拟流水线上的工人，
 工人一直在流水线上作业，零件（可看作客户端发的请求）一到达，工人就开始进行工作，无零件时工作处于等待状态
具体的业务代码有详注释：


一测试类：
package worker.thread.pattern;

public class Main {
    public static void main(String[] args) {
    	 Channel channel = new Channel(5);   // Worker Thread数量
         channel.startWorkers();
         ClientThread alice = new ClientThread("Alice", channel);
         ClientThread bobby = new ClientThread("Bobby", channel);
         ClientThread chris = new ClientThread("Chris", channel);
         alice.start();
         bobby.start();
         chris.start();

         try {
             Thread.sleep(5000);
         } catch (InterruptedException e) {
         }
         alice.stopThread();
         bobby.stopThread();
         chris.stopThread();
         channel.stopAllWorkers();
    }
}


2.客户端类：主要发送工作请求
package worker.thread.pattern;

import java.util.Random;

/**
 * 不停的发送请求给工人类
 * 
 * 1.创建请求实例，并将实例发送给管理工人线程的类
 * 
 * @author lxb
 *
 */
public class ClientThread extends Thread {

	private final Channel channel;
	private static final Random random = new Random();

	private volatile boolean terminated = false; // 停止请求标志

	public ClientThread(String name, Channel channel) {
		super(name);
		this.channel = channel;
	}

	public void run() {
		try {
			for (int i = 0; !terminated; i++) {
				try {
					Request request = new Request(getName(), i);
					channel.putRequest(request);
					Thread.sleep(random.nextInt(1000));
				} catch (InterruptedException e) {
					terminated = true;
				}
			}
		} finally {
			System.out.println(Thread.currentThread().getName()
					+ " is terminated.");
		}
	}

	public void stopThread() {
		terminated = true;
		interrupt();
	}
}



3.请求对象：

package worker.thread.pattern;

import java.util.Random;

/**
 * 请求实体类
 * @author lxb
 *
 */
public class Request {

	private final String name; //  
    private final int number;  // 
    private static final Random random = new Random();
    public Request(String name, int number) {
        this.name = name;
        this.number = number;
    }
    public void execute() {
        System.out.println(Thread.currentThread().getName() + " executes " + this);
        try {
            Thread.sleep(random.nextInt(1000));
        } catch (InterruptedException e) {
        }
    }
    public String toString() {
        return "[ Request from " + name + " No." + number + " ]";
    }
}


4.请求队列与工人线程管理类：
package worker.thread.pattern;

/**
 * 用来接受客户端线程发来的请求、将请求传递给工人线程，所以至少需要以下数据结构：
 * 1.一个队列来保存客户端发来的请求，还必须提供两个接口，一个putRequest保存请求， 另一个是取出请求takeRequest
 * 2.线程池来管理各个工人线程,这个池可以是一个线程数组，数组大小可自定义
 * 
 * @author lxb
 *
 */
public class Channel {

	private static final int MAX_REQUEST = 100; // 假设最大的请求队列值

	private final Request[] requestQueue; // 保存的请求队列

	private int tail;
	private int head;
	private int count; // 请求队列中的请求数量

	private final WorkerThread[] threadPool; // 工人线程池

	public Channel(int threads) {

		this.requestQueue = new Request[MAX_REQUEST];
		this.head = 0;
		this.count = 0;
		this.tail = 0;
		threadPool = new WorkerThread[threads];

		for (int i = 0; i < threadPool.length; i++) {
			 threadPool[i] = new WorkerThread("Worker-" + i, this);
		}
	}
	
	/**
	 * 开启工作线程
	 */
	public void startWorkers() {
		for (int i = 0; i < threadPool.length; i++) {
			threadPool[i].start();
		}
	}
	
	public void stopAllWorkers(){
		for(int i=0;i<threadPool.length;i++){
			threadPool[i].stopThread();
		}
	}

	/**
	 * 保存客户端发来的工作请求
	 * 
	 * @param request
	 * @throws InterruptedException 
	 */
	public synchronized void putRequest(Request request) throws InterruptedException {
		while (count >= requestQueue.length) {
			wait();
		}

		requestQueue[tail] = request;

		tail = (tail + 1) % requestQueue.length;

		count++;

		notifyAll();
	}

	public synchronized Request takeRequest() throws InterruptedException {
		while (count <= 0) {
			wait();
		}

		Request request = requestQueue[head];
		head = (head + 1) % requestQueue.length;

		count--;
		notifyAll();
		return request;
	}

}


5工人线程：


package worker.thread.pattern;

/**
 * 工作线程，处理客户端发来的具体的工作请求
 * @author lxb
 *
 */
public class WorkerThread extends Thread{

	private final Channel channel;
	private volatile boolean terminated = false;	//停止请求标志
	
	public WorkerThread(String name,Channel channel){
		super(name);
		this.channel = channel;
	}
	
	public void run() {
        try {
            while (!terminated) {
                try {                                           
                    Request request = channel.takeRequest();    
                    request.execute();                          
                } catch (InterruptedException e) {
                    terminated = true;
                }                                               
            }
        } finally {
            System.out.println(Thread.currentThread().getName() + " is terminated.");
        }
    }
    public void stopThread() {
        terminated = true;
        interrupt();
    }
}