go并发编程题目

交替打印1-100的奇数偶数

var (
	POOL = 100
)

func groutine1(dep, nxt chan int) {
	for i := 1; i <= POOL; i += 2 {
		<-dep
		fmt.Println("groutine-1:", i)
		nxt <- 1
	}
}
func groutine2(dep, nxt chan int) {
	for i := 2; i <= POOL; i += 2 {
		<-dep
		fmt.Println("groutine-2:", i)
		nxt <- 1
	}
}
func main() {
	c1 := make(chan int)
	c2 := make(chan int)
	go groutine1(c1, c2)
	go groutine2(c2, c1)
	c1 <- 1
	time.Sleep(time.Second * 1)
}

用chan 信号通知另一个goroutine执行。

用chan 信号阻塞等待通知

交替打印1234

极客上一道有意思的题，假设有4个 goroutine，编号为1，2，3，4。每秒钟会有一个 goroutine 打印出它自己的编号。现在让你写一个程序，要求输出的编号总是按照1，2，3，4这样的顺序打印。类似下图，

package main

import (
  "fmt"
  "time"
)

type token struct{}

func main() {
  num := 4
  var chs []chan token
  // 4 个work
  for i := 0; i < num; i++ {
    chs = append(chs, make(chan token))
  }
  for j := 0; j < num; j++ {
    go worker(j, chs[j], chs[(j+1)%num])
  }
  // 先把令牌交给第一个
  chs[0] <- struct{}{}
  select {}
}

func worker(id int, ch chan token, next chan token) {
  for {
    // 对应work 取得令牌
    token := <-ch
    fmt.Println(id + 1)
    time.Sleep(1 * time.Second)
    // 传递给下一个
    next <- token
  }
}

用chan实现互斥锁

var num int
func add(h chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    h <- 1
    num += 1
    <-h
}
func main() {
    ch := make(chan int, 1)
    wg := &sync.WaitGroup{}
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go add(ch, wg)
    }
    wg.Wait()
    fmt.Println("num:", num)
}

一堆任务，限制并发度为3

利用chan缓冲区

// main_chan.go
func main() {
	var wg sync.WaitGroup
	ch := make(chan struct{}, 3)
	for i := 0; i < 10; i++ {
		ch <- struct{}{} //写
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			log.Println(i)
			time.Sleep(time.Second)
			<-ch
		}(i)
	}
	wg.Wait()
}

make(chan struct{}, 3) 创建缓冲区大小为 3 的 channel，在没有被接收的情况下，至多发送 3 个消息则被阻塞。
开启协程前，调用 ch <- struct{}{}，若缓存区满，则阻塞。
协程任务结束，调用 <-ch 释放缓冲区。
sync.WaitGroup 并不是必须的

非阻塞的尝试读chan或者写chan

可能会因为阻塞而不成功，如果阻塞了就不会读写成功

非阻塞尝试下写

select {
	case done <- true:
	default:
		return
}

非阻塞尝试下读

select {
	case <- done:
	default:
		return
}

无用测试code

func TestB(t *testing.T) {
	a := make(chan int)
	go func() {
		x := <-a
		fmt.Println(x)
	}()
	select {
	case a <- 1:
	default:
		fmt.Println("写失败")
	}
	select {
	case x := <- a:
		fmt.Println(x)
	default:
		fmt.Println("读失败")
	}
	time.Sleep(1 * time.Second)
}

超时控制

可以用time.After

select {
  case xxx...
  
	case <-time.After(time.Millisecond):
		return fmt.Errorf("timeout")
}

定时器

	for x := range time.Tick(time.Second){
		fmt.Println(x)
	}

实现一个可重入锁

参考

重入锁就是一个user可以多次Lock 最后UnLock和Lock数量一致就解锁。可重入锁又称为递归锁，是指在同一个线程在外层方法获取锁的时候，在进入该线程的内层方法时会自动获取锁，不会因为之前已经获取过还没释放而阻塞。

实现一个可重入锁需要这两点：

记住持有锁的线程，比如user
统计重入的次数, 比如times，如果是同user，那么times不会有并发访问的

package main

import (
	"fmt"
	"runtime"
	"sync"
)

type ReentrantLock struct {
	mu       sync.Mutex
	owner    int          // 当前持有锁的 Goroutine ID
	recursion int          // 当前 Goroutine 持有锁的次数
}

func (rl *ReentrantLock) Lock() {
	gid := getGoroutineID() // 获取当前 Goroutine 的 ID
	rl.mu.Lock()
	defer rl.mu.Unlock()

	if rl.owner == gid {
		// 如果当前 Goroutine 已持有锁，增加递归计数
		rl.recursion++
		return
	}

	// 等待锁被释放，成为新的锁持有者
	for rl.recursion > 0 {
		rl.mu.Unlock()
		rl.mu.Lock()
	}

	rl.owner = gid
	rl.recursion = 1
}

func (rl *ReentrantLock) Unlock() {
	gid := getGoroutineID()
	rl.mu.Lock()
	defer rl.mu.Unlock()

	if rl.owner != gid {
		panic("unlock of unowned lock")
	}

	rl.recursion--
	if rl.recursion == 0 {
		// 如果递归计数为 0，释放锁
		rl.owner = 0
	}
}

// 获取 Goroutine ID 的辅助函数
func getGoroutineID() int {
	buf := make([]byte, 64)
	n := runtime.Stack(buf, false)
	var id int
	fmt.Sscanf(string(buf[:n]), "goroutine %d ", &id)
	return id
}

func main() {
	rl := &ReentrantLock{}

	var wg sync.WaitGroup
	wg.Add(1)

	go func() {
		defer wg.Done()
		fmt.Println("Goroutine 1 trying to acquire lock...")
		rl.Lock()
		fmt.Println("Goroutine 1 acquired lock")

		fmt.Println("Goroutine 1 trying to re-acquire lock...")
		rl.Lock()
		fmt.Println("Goroutine 1 re-acquired lock")

		fmt.Println("Goroutine 1 releasing lock...")
		rl.Unlock()
		fmt.Println("Goroutine 1 released one lock level")

		rl.Unlock()
		fmt.Println("Goroutine 1 fully released lock")
	}()

	wg.Wait()
}

检查times为0这个for循环有点挫，更好的是使用信号量。

type ReentrantLock struct {
    lock *sync.Mutex
    cond *sync.Cond
    recursion int32
    host     int64
}

func (rt *ReentrantLock) Lock()  {
    id := GetGoroutineID()
    rt.lock.Lock()
    defer rt.lock.Unlock()

    if rt.host == id{
        rt.recursion++
        return
    }

    for rt.recursion != 0{
        rt.cond.Wait()
    }
    rt.host = id
    rt.recursion = 1
}

func (rt *ReentrantLock) Unlock()  {
    rt.lock.Lock()
    defer rt.lock.Unlock()

    if rt.recursion == 0 || rt.host != GetGoroutineID() {
        panic(fmt.Sprintf("the wrong call host: (%d); current_id: %d; recursion: %d", rt.host,GetGoroutineID(),rt.recursion))
    }

    rt.recursion--
    if rt.recursion == 0{
        rt.cond.Signal()
    }
}

Go语言的发明者认为，如果当你的代码需要重入锁时，那就说明你的代码有问题了，我们正常写代码时，从入口函数开始，执行的层次都是一层层往下的，如果有一个锁需要共享给几个函数，那么就在调用这几个函数的上面，直接加上互斥锁就好了，不需要在每一个函数里面都添加锁，再去释放锁。

实现一个读写锁

参考

实现连接池

在 Go 语言中，实现一个基本的 连接池 是一个常见的需求，尤其是在处理数据库连接、HTTP 请求连接或其他类型的网络连接时。连接池的主要作用是复用已有的连接，避免频繁的建立和销毁连接，从而提高性能。

channel存连接，并发读和取。select试探是否有空闲的。
api: 取, 还

// Connection 模拟一个连接对象
type Connection struct {
	ID int
}

// ConnectionPool 连接池结构
type ConnectionPool struct {
	pool     chan *Connection // 存放连接的 channel
	maxConns int              // 最大连接数
	mu       sync.Mutex       // 保护连接池的并发访问, 创建时用到.
}

// NewConnectionPool 创建一个新的连接池
func NewConnectionPool(maxConns int) *ConnectionPool {
	return &ConnectionPool{
		pool:     make(chan *Connection, maxConns), // 使用缓冲 channel 存放连接
		maxConns: maxConns,
	}
}

// GetConnection 从连接池中获取一个连接
func (p *ConnectionPool) GetConnection() (*Connection, error) {
	select {
	case conn := <-p.pool: // 从池中取出连接
		fmt.Println("Reusing connection:", conn.ID)
		return conn, nil
	default:
		// 如果没有空闲连接，可以选择创建新的连接
		return p.createConnection(), nil
	}
}

// createConnection 创建一个新的连接
func (p *ConnectionPool) createConnection() *Connection {
	p.mu.Lock()
	defer p.mu.Unlock()

	// 如果连接池没有达到最大连接数，创建新的连接
	if len(p.pool) < p.maxConns {
		conn := &Connection{ID: len(p.pool) + 1}
		fmt.Println("Creating new connection:", conn.ID)
		return conn
	}

	// 如果连接池已满，返回 nil 或报错
	return nil
}

// ReturnConnection 将连接放回连接池
func (p *ConnectionPool) ReturnConnection(conn *Connection) {
	select {
	case p.pool <- conn: // 将连接放回池中
		fmt.Println("Returning connection:", conn.ID)
	default:
		// 如果池已满，则丢弃连接
		fmt.Println("Connection pool is full, discarding connection:", conn.ID)
	}
}

// 关闭连接池，清理所有连接
func (p *ConnectionPool) Close() {
	close(p.pool)
}

实现协程池

线程池原理: 在线程池中存在几个概念：核心线程数、最大线程数、任务队列。核心线程数指的是线程池的基本大小；最大线程数指的是，同一时刻线程池中线程的数量最大不能超过该值；任务队列是当任务较多时，线程池中线程的数量已经达到了核心线程数，这时候就是用任务队列来存储我们提交的任务。与其他池化技术不同的是，线程池是基于生产者-消费者模式来实现的，任务的提交方是生产者，线程池是消费者。当我们需要执行某个任务时，只需要把任务扔到线程池中即可。

// Task 定义任务的结构体
type Task struct {
	ID int
  handle func(){}
}

// GoroutinePool 定义协程池的结构体
type GoroutinePool struct {
	taskQueue chan Task   // 存放任务的队列
	wg        sync.WaitGroup // 等待所有任务完成
	maxGoroutines int      // 最大协程数
}

// NewGoroutinePool 创建一个新的协程池
func NewGoroutinePool(maxGoroutines int) *GoroutinePool {
	return &GoroutinePool{
		taskQueue:    make(chan Task, maxGoroutines), // 任务队列的大小与最大协程数相同
		maxGoroutines: maxGoroutines,
	}
}

// Run 执行协程池
func (pool *GoroutinePool) Run() {
	for i := 0; i < pool.maxGoroutines; i++ {
		go pool.worker(i) // 启动 maxGoroutines 个协程
	}
}

// worker 执行任务的工作函数
func (pool *GoroutinePool) worker(id int) {
	for task := range pool.taskQueue {
		fmt.Printf("Worker %d started task %d\n", id, task.ID)
		// 模拟任务处理时间
		time.Sleep(time.Second)
		fmt.Printf("Worker %d finished task %d\n", id, task.ID)
		pool.wg.Done() // 标记任务完成
	}
}

// AddTask 向协程池中添加任务
func (pool *GoroutinePool) AddTask(task Task) {
	pool.wg.Add(1)        // 添加一个任务，等待协程完成
	pool.taskQueue <- task // 将任务加入任务队列
}

// Wait 等待所有任务完成
func (pool *GoroutinePool) Wait() {
	pool.wg.Wait() // 等待所有任务完成
	close(pool.taskQueue) // 关闭任务队列
}

func main() {
	// 创建一个最大协程数为 3 的协程池
	pool := NewGoroutinePool(3)

	// 启动协程池
	pool.Run()

	// 向池中添加 10 个任务
	for i := 1; i <= 10; i++ {
		task := Task{ID: i}
		pool.AddTask(task)
	}

	// 等待所有任务完成
	pool.Wait()

	fmt.Println("All tasks have been completed.")
}

实现限流器

参考系统设计里的限流器设计。

不能用atomic，实际只能用mutex。

参考个人博客: https://mafulong.eu.org/2021/04/12/%E9%99%90%E6%B5%81%E5%99%A8/#golang%E5%AE%9E%E7%8E%B0%E4%BB%A4%E7%89%8C%E6%A1%B6–%E9%98%BB%E5%A1%9Efifo

实现熔断器

个人blog

实现阻塞读且并发安全的map

GO里面MAP如何实现key不存在 get操作等待直到key存在或者超时，保证并发安全，且需要实现以下接口：

type sp interface {
    Out(key string, val interface{})  //存入key /val，如果该key读取的goroutine挂起，则唤醒。此方法不会阻塞，时刻都可以立即执行并返回
    Rd(key string, timeout time.Duration) interface{}  //读取一个key，如果key不存在阻塞，等待key存在或者超时
}

解析：

看到阻塞协程第一个想到的就是channel，题目中要求并发安全，那么必须用锁，还要实现多个goroutine 读的时候如果值不存在则阻塞，直到写入值，那么每个键值需要有一个阻塞goroutine 的 channel。

type Map struct {
	c   map[string]*entry
	rmx *sync.RWMutex
}
type entry struct {
	ch      chan struct{}
	value   interface{}
	isExist bool
}

func (m *Map) Out(key string, val interface{}) {
	m.rmx.Lock()
	defer m.rmx.Unlock()
	item, ok := m.c[key]
	if !ok {
		m.c[key] = &entry{
			value: val,
			isExist: true,
		}
		return
	}
	item.value = val
	if !item.isExist {
		if item.ch != nil {
			close(item.ch)
			item.ch = nil
		}
	}
	return
}

Fulongのblog

go并发编程题目