GO语言入门第五节 Go语言的并发编程

时间：2022-12-19 19:06:22

写在前面

因为并发相关的东西又多又长。。所以这个专题会分成多篇博客来写啦。。
本篇文章包括

  • 携程机制，携程和线程的区别
  • 使用锁来控制并发
  • 使用通道（channel）来控制并发
  • 通道的多路控制和超时（select语句块）

Go语言的并发操作（一）

1. Go语言的携程机制

  • 线程和携程的区别（携程是更加轻量级的线程）
  • JDK5之后一个线程的默认栈道大小是5M，而携程栈的大小是2K
  • Java中线程和系统线程的对应关系是1:1，但是携程是n:m

线程由于涉及到处理器切换人物，会导致吞吐率下降。而使用携程，可以做到多个携程只会使用一个系统现场。利用这种方式降低切换线程上下文的成本

  • 携程的简单使用
    携程的使用非常简单，在需要运行的函数前面加上​​go​​关键字就可以了
    个人认为只需要把携程看成轻量级线程，使用时可以直接当成线程使用

GO语言入门第五节 Go语言的并发编程

上面的程序输出结果如下（由于处理机调用的时间不一致，所以并不会出现顺序的数字）

=== RUN TestGroutine
4
7
5
6
1
8
9
0
2
3
--- PASS: TestGroutine (0.10s)
PASS

2. Go 语言的共享内存并发机制

  • 使用锁来控制多线程程序
    先看一个在没有锁的情况下使用携程（多线程）把一个值累加点结果

GO语言入门第五节 Go语言的并发编程



结果如下，可以看见我们丢失了正确的鞋操作


counter =  4799


为了获得正确的结果，我们需要使用锁，和等待队列


GO语言入门第五节 Go语言的并发编程


3. CSP的并发控制


CSP相当于是通过通道（发送消息）的机制来进行线程通讯。CSP的操作是通过一个Channel来完成的。也就是，这种方式会比使用直接通讯耦合度更低


  • Channel的交互机制


    

  • 主动式交互：如果消息消费者离线，那么消息生产者就会阻塞
    • 

  • 队列（缓存）式交互：在消息队列未满之前，消息的生产者可以一直放入消息，如果消息队列满了，那么消息生产者就会阻塞
    • 



  • CSP实现异步调用，延迟返回值


  • 例如这里的Service方法是需要异步调用的


//这个方法做一些耗时的操作，需要异步调用，
func service() string {
time.Sleep(time.Second*1)
return "do something"
}


  • 包装上面的方法，返回一个channel（注意代码里的chan类型的定义方式和把函数放入调用的方式）


//使用一个异步调用包装上面的方法
func asyncService() chan string {
//创建一个string类型的频道
returnChannel := make(chan string)
//异步调用
go func() {
fmt.Println("开始异步调用")
returnChannel <- service()
fmt.Println("异步调用结束")
  }()
return returnChannel
}


  • 这样调用​​asyncService​​方法的时候虽然可以很快的返回chan类型的数据，但是本质上任然是通过携程异步调用的，立即去获取chan里的值将会因为异步调用没有返回值而被阻塞
    我给出一个调用案例如下


//在程序入口的线程里也要做一些操作
func TestTask(t *testing.T)  {
//开始异步调用并且等待结果
result := asyncService()

//开始做别的事情
time.Sleep(500*time.Millisecond)
fmt.Println("do something else")

//这里再等待结果
fmt.Println(<-result)
}


结果如下


=== RUN   TestTask

开始异步调用

do something else

异步调用结束

do something

--- PASS: TestTask (1.01s)

PASS


  • 创建一个带消息容量的channel是需要再make方法的参数后面加上一个int来表示消息队列的容量就好了，如下


returnChannel := make(chan string,1)




由于我对我的老本行——java后端的异步调用还不是很熟悉，那我就趁热打铁，写一个Demo实现Java的异步调用


这一部分是使用Java的Callable和FutureTask来完成异步调用，在获取调用结果的时候如果没有完成，主线程就会阻塞直到任务完成。如果不感兴趣可以直接跳到下一个大标题


  • 使用Callable和ExecutorService线程池来完成任务


package com.libi.callable;

import java.util.concurrent.*;

/**
* @author :Libi
* @version :1.0
* @date :2019-06-17 19:39
* 用于了解Callable的使用方法
*/
public class CallableDemo implements Callable<String> {

/**
* 这个方法是用于异步调用的
* @return
* @throws Exception
*/
@Override
public String call() throws Exception {
System.out.println("do something in 5 seconds");
Thread.sleep(5000);
return "完成，这个任务做完了";
    }

/**
* 这个线程池用于调用callable任务，并且可以获得
*/
private ExecutorService executorService = Executors.newSingleThreadExecutor();

public static void main(String[] args) throws ExecutionException, InterruptedException {
CallableDemo callableDemo = new CallableDemo();
System.out.println("开始异步调用");
//这一步可以封装成方法
Future<String> submit = callableDemo.executorService.submit(callableDemo);
String s = submit.get();
System.out.println(s);
callableDemo.executorService.shutdown();
    }
}


  • 使用FutureTask的普通的Thread来完成任务


package com.libi.task;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

/**
* @author :Libi
* @version :1.0
* @date :2019-06-17 19:51
* 这个类是研究FutureTask的使用方法而创建的类
*/
public class TaskDemo {

/**
* 这种方法就没有用到线程池，但是也简单的实现类异步调用的
* @return
*/
private FutureTask<String> asyncService() {
FutureTask<String> stringFutureTask = new FutureTask<String>(() -> {
Thread.sleep(5000);
return "任务完成";
        });
new Thread(stringFutureTask).start();
return stringFutureTask;
    }

public static void main(String[] args) throws ExecutionException, InterruptedException {
TaskDemo demo = new TaskDemo();
System.out.println("开始异步调用");
FutureTask<String> stringFutureTask = demo.asyncService();
String s = stringFutureTask.get();
System.out.println("异步调用结束");
System.out.println(s);
    }
}


4. 多路选择和超时（select块的使用）


select块就是用于多个异步调用的多路选择和超时控制。


select语句块类似switch，每一个case里都要使用一个频道（chan类型）来获得数据。只要有一个channel返回了数据，那么这个channel的语句块就会执行。如果都没有返回值，有default语句块的话就会执行default语句块


这里的channel应该提前启动好，当我们要获取结果时再去做相关处理





func TestSelect1(t *testing.T) {
s := asyncService()
time.Sleep(time.Millisecond*1000)
select {
//这一个语句块是为了做超时处理，10s后如果没有结果他就会返回结果
//（当然有了default语句块这个语句块也就没有意义了）
case <-time.After(10*time.Second):
print("10s")
case ret := <-s:
print("result:",ret)
default:
t.Error("error")
   }
}


标签：
相关文章