协程（又名纤程），轻量级线程（建立在线程基础上，属于用户态调用），非阻塞式编程（像同步编写一样），在用户态内进行任务调度，避免与内核态过多交互问题，提高程序快速响应。协程使用挂起当前上下文替代阻塞，被挂起后的协程可以去运行其它active task,即协程可以被复用，相比于线程，减少了线程资源的大量浪费。
备注
挂起：保存当前运行状态，释放资源，此时协程可去做其它工作，可充分利用资源
阻塞：占用资源未释放，等待状态
基本使用：
fun runAsync()= runBlocking {
val time = measureTimeMillis {//系统函数统计时间
val one = async { doSomethingUsefulOne() }//异步调用，返回结果
val two = async { doSomethingUsefulTwo() }
println("The answer is ${one.await() + two.await()}")//等待异步执行完成（await调用会挂起当前线程,等待执行结果完成后，通过调用resume恢复挂起前状态）
}
println("Completed in $time ms")
}
//协程coroutines 调用的方法需要用suspend修饰，告诉编译器此函数可以被挂起
suspend fun doSomethingUsefulOne(): Int {
delay(1000L) // pretend we are doing something useful here
return 13
}
suspend fun doSomethingUsefulTwo(): Int {
delay(1000L) // pretend we are doing something useful here, too
return 29
}
这里面没有使用异步+回调，直接像写同步代码一样，简洁
launch 异步执行没有返回结果，产生Job对象用于cancel，join处理
fun cancelCoroutine() = runBlocking {
val startTime = System.currentTimeMillis()
val job = launch(Dispatchers.Default) {
var nextPrintTime = startTime
var i = 0
while (isActive) { // cancellable computation loop
// print a message twice a second
if (System.currentTimeMillis() >= nextPrintTime) {
println("job: I'm sleeping ${i++} ...")
nextPrintTime += 500L
}
}
}
delay(1300L) // delay a bit
println("main: I'm tired of waiting!")
job.cancelAndJoin() // cancels the job and waits for its completion
println("main: Now I can quit.")
}
线程之间切换，使用withContext
fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
fun jumpCor(){//创建单线程coroutines
newSingleThreadContext("Ctx1").use { ctx1 ->
newSingleThreadContext("Ctx2").use { ctx2 ->
runBlocking(ctx1) {
log("Started in ctx1")
withContext(ctx2) {
log("Working in ctx2")
}
log("Back to ctx1")
}
}
}
}
协程必须关联CoroutineScope以便于管理追踪，方法内创建Scope
suspend fun showSomeData() = coroutineScope {//此处coroutineScope属于out scope的child scop
val data = async(Dispatchers.IO) { // IO task io线程调用操作
// ... load some UI data for the Main thread ...
}
withContext(Dispatchers.Main){//UI task UI更新
val result = data.await()
// display(result)
}
}
协程上下文环境，CoroutineScope，CoroutineContext
每个协程运行需要在指定Scope内才能使用协程相关方法delay,asyc,launch，创建CoroutineScope ，runBlocking函数内部会创建CoroutineScope，系统提供GlobalScope,MainScope等辅助类创建Scope
也可以通过CoroutineContext和Job创建自己的CoroutineScope
fun sampleCreateCorountine(){
//create corountine scope
//自定义CoroutineScope
val coroutineContext = Dispatchers.Default
val job = Job()
val coroutineScope = CoroutineScope(coroutineContext + job)
//创建child scope
coroutineScope.launch {
}
//创建全局Scope
GlobalScope.launch (Dispatchers.Default+CoroutineName("global background thread")){
}
//创建主线程分发处理Scope
MainScope().launch {
}
}
类内部定义协程
1，直接继承CoroutineScope
class SomethingWithLifecycle : CoroutineScope {
// 使用job来管理你的SomethingWithLifecycle的所有子协程
private val job = Job()
override val coroutineContext: CoroutineContext
get() = Dispatchers.Main + job
fun destory(){//退出取消
job.cancel()
}
}
2，直接使用已定义Scope
class CorMyActivity : AppCompatActivity(), CoroutineScope by MainScope() {
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
showSomeData()
}
/**
* Note how coroutine builders are scoped: if activity is destroyed or any of the launched coroutines
in this method throws an exception, then all nested coroutines are cancelled.
*/
fun showSomeData() = launch {
// <- extension on current activity, launched in the main thread
// ... here we can use suspending functions or coroutine builders with other dispatchers
// draw(data) // draw in the main thread
}
override fun onDestroy() {
super.onDestroy()
cancel()
}
}
Dispatchers，协程分发器：
fun dispatchTask()= runBlocking<Unit> {
// it inherits the context (and thus dispatcher) from the CoroutineScope that it is being launched from.
launch { // context of the parent, main runBlocking coroutine
println("main runBlocking : I'm working in thread ${Thread.currentThread().name}")
}
//执行coroutine是在调用者的线程，但是当在coroutine中第一个挂起之后，后面所在的线程将完全取决于
// 调用挂起方法的线程(如delay一般是由kotlinx.coroutines.DefaultExecutor中的线程调用)
//Unconfined在挂起后在delay的调用线程DefaultExecutor执行
launch(context = Dispatchers.Unconfined) { // not confined -- will work with main thread
println("Unconfined : I'm working in thread ${Thread.currentThread().name}")
}
// coroutines are launched in GlobalScope,uses shared background pool of threads
//uses the same dispatcher as GlobalScope.launch
//Dispatchers.Default 处理cup密集型任务，线程数为cpu内核数，最少为2，Dispatchers.IO 处理阻塞性IO，socket密集度任务，数量随任务多少变化，默认最大数量64
launch(context = Dispatchers.Default) { // will get dispatched to DefaultDispatcher
println("Default : I'm working in thread ${Thread.currentThread().name}")
}
//creates a thread for the coroutine to run
launch(newSingleThreadContext("MyOwnThread")) { // will get its own new thread
println("newSingleThreadContext: I'm working in thread ${Thread.currentThread().name}")
}
}
suspend 是如何工作的?
Kotlin 使用堆栈帧来管理要运行哪个函数以及所有局部变量。暂停协程时，
会复制并保存当前的堆栈帧以供稍后使用。恢复协程时，调度器会将堆栈帧从其保存位置复制回来，然后函数再次开始运行
协程间通信之channel
协程之间通过channel进行数据传递，生产者->消费者模式
例：
fun channelTest()= runBlocking {
val channel = Channel<Int>()
launch {//生产数据
for (x in 1..5) channel.send(x * x)
channel.close() //关闭停止
}
// 循环接收直到channnel close
for (y in channel) println(y)
println("Done!")
}
生产者每生产一个数据就发送到channel里，消费者等待接收数据，
channel分类：
SendChannel：创建的producers类型属于sendChannel实例
ReceiveChannel：创建的consumers类型属于receiveChannel实例
Channel：继承SendChannel和ReceiveChannel即可send，又可以receive数据
channel类型：
Unlimited channel：容量无限制，producer不断生产数据，可能会产生OutOfMemoryException，consumer接收数据时，如果channel内数据为空则会挂起
Buffered channel：指定 channel size，当生产者的数据达到buffer size大小则send会挂起，直到channel内数据量小于size才能继续生产数据
Rendezvous：是bufferred channel size=0，当producer生成数据send时如果没有consumer接受，则producer会挂起直到consumer取走数据，才继续send下一个数据，即实现同步传递数据功能
Conflated channel：producer不停地send数据，后面的数据会覆盖前面已经存在的数据，consumer始终取到最新的数据
val rendezvousChannel = Channel<String>()//同步传递
val bufferedChannel = Channel<String>(10)//指定size pool
val conflatedChannel = Channel<String>(Channel.CONFLATED)//channel内数据实时更新
val unlimitedChannel = Channel<String>(Channel.UNLIMITED)//无容量限制
协程结合Architecture ViewModel使用
class NewsViewModel: ViewModel() {
private val mApi:WebServer
init {
mApi = WebServer()
}
val dataNews: MutableLiveData<DataResource<NewsDataRsp>> by lazy {
// MutableLiveData<DataResource<NewsDataRsp>>().also {
// loadNewsData(minId=null)
// }
MutableLiveData<DataResource<NewsDataRsp>>()
}
fun loadNewsData(pageIndex:Int =1,countItem:Int = 20,minId:String?=null){
runCoroutine(dataNews){
val mp = mutableMapOf("encode" to "ywjh","source" to "app","sys" to "android","banner" to "banner",
"limit" to countItem.toString(),"version" to "7002000")
if(pageIndex>1 && false==minId.isNullOrEmpty()){
mp.put("min_id",minId)
}
val response = mApi.commonDataSourceApi.getNewsData(mp).execute()
return@runCoroutine response.body()!!
}
}
fun fetchNews(pageIndex:Int =1,countItem:Int = 20,minId:String){
val mp = mutableMapOf("encode" to "ywjh","source" to "app","sys" to "android","banner" to "banner",
"limit" to countItem.toString(),"version" to "7002000")
if(pageIndex>1 && false==minId.isNullOrEmpty()){
mp.put("min_id",minId)
}
val cor = CoroutineScope(Dispatchers.IO)
cor.launch {
try {
val response = mApi.commonDataSourceApi.getNewsData(mp).execute()
dataNews.postValue(DataResource(DataResource.Status.COMPLETED, response.body(), null))
} catch (exception: Exception) {
dataNews.postValue(DataResource(DataResource.Status.COMPLETED, null, exception))
}
}
}
suspend fun simpleGetData(pageIndex:Int =1,countItem:Int = 20,minId:String) = withContext(Dispatchers.IO) {
val mp = mutableMapOf("encode" to "ywjh","source" to "app","sys" to "android","banner" to "banner",
"limit" to countItem.toString(),"version" to "7002000")
if(pageIndex>1 && false==minId.isNullOrEmpty()){
mp.put("min_id",minId)
}
try {
val response = mApi.commonDataSourceApi.getNewsData(mp).execute()
dataNews.postValue(DataResource(DataResource.Status.COMPLETED, response.body(), null))
} catch (exception: Exception) {
dataNews.postValue(DataResource(DataResource.Status.COMPLETED, null, exception))
}
}
private fun <T> runCoroutine(correspondenceLiveData: MutableLiveData<DataResource<T>>, block: suspend () -> T) {
correspondenceLiveData.value = DataResource(DataResource.Status.LOADING, null, null)
GlobalScope.launch(Dispatchers.IO) {
try {
val result = block()
correspondenceLiveData.postValue(DataResource(DataResource.Status.COMPLETED, result, null))
} catch (exception: Exception) {
// val error = ErrorConverter.convertError(exception)
correspondenceLiveData.postValue(DataResource(DataResource.Status.COMPLETED, null, exception))
}
}
}
}
