1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

type Cond struct {

   noCopy noCopy

   // 锁

   L Locker

   // 等待队列

   notify  notifyList

   checker copyChecker

}

func NeWCond(l Locker) *Cond

func (c *Cond) Broadcast()

func (c *Cond) Signal()

func (c *Cond) Wait()

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

package main

import (

   "container/list"

   "fmt"

   "math/rand"

   "sync"

   "time"

)

// 连接池

type Pool struct {

   lock    sync.Mutex // 锁

   clients list.List  // 连接

   cond    *sync.Cond // cond实例

   close   bool       // 是否关闭

}

// Redis Client

type Client struct {

   id int32

}

// 创建Redis Client

func NewClient() *Client {

   return &Client{

      id: rand.Int31n(100000),

   }

}

// 关闭Redis Client

func (this *Client) Close() {

   fmt.Printf("Client:%d 正在关闭", this.id)

}

// 创建连接池

func NewPool(maxConnNum int) *Pool {

   pool := new(Pool)

   pool.cond = sync.NewCond(&pool.lock)

   // 创建连接

   for i := 0; i < maxConnNum; i++ {

      client := NewClient()

      pool.clients.PushBack(client)

   }

   return pool

}

// 从池子中获取连接

func (this *Pool) Pull() *Client {

   this.lock.Lock()

   defer this.lock.Unlock()

   // 已关闭

   if this.close {

      fmt.Println("Pool is closed")

      return nil

   }

   // 如果连接池没有连接 需要阻塞

   for this.clients.Len() <= 0 {

      this.cond.Wait()

   }

   // 从链表中取出头节点，删除并返回

   ele := this.clients.Remove(this.clients.Front())

   return ele.(*Client)

}

// 将连接放回池子

func (this *Pool) Push(client *Client) {

   this.lock.Lock()

   defer this.lock.Unlock()

   if this.close {

      fmt.Println("Pool is closed")

      return

   }

   // 向链表尾部插入一个连接

   this.clients.PushBack(client)

   // 唤醒一个正在等待的goruntine

   this.cond.Signal()

}

// 关闭池子

func (this *Pool) Close() {

   this.lock.Lock()

   defer this.lock.Unlock()

   // 关闭连接

   for e := this.clients.Front(); e != nil; e = e.Next() {

      client := e.Value.(*Client)

      client.Close()

   }

   // 重置数据

   this.close = true

   this.clients.Init()

}

func main() {

   var wg sync.WaitGroup

   pool := NewPool(3)

   for i := 1; i <= 10; i++ {

      wg.Add(1)

      go func(index int) {

         defer wg.Done()

         // 获取一个连接

         client := pool.Pull()

         fmt.Printf("Time:%s | 【goruntine#%d】获取到client[%d]\n", time.Now().Format("15:04:05"), index, client.id)

         time.Sleep(time.Second * 5)

         fmt.Printf("Time:%s | 【goruntine#%d】使用完毕，将client[%d]放回池子\n", time.Now().Format("15:04:05"), index, client.id)

         // 将连接放回池子

         pool.Push(client)

      }(i)

   }

   wg.Wait()

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

//

//    c.L.Lock()

//    for !condition() {

//        c.Wait()

//    }

//    ... make use of condition ...

//    c.L.Unlock()

//

func (c *Cond) Wait() {

   c.checker.check()

   // 添加到等待队列

   t := runtime_notifyListAdd(&c.notify)

   c.L.Unlock()

   // 阻塞

   runtime_notifyListWait(&c.notify, t)

   c.L.Lock()

}

1

2

3

4

5

6

7

8

// 如果连接池没有连接 需要阻塞

for this.clients.Len() <= 0 {

  this.cond.Wait()

}

// 获取连接

ele := this.clients.Remove(this.clients.Front())

return ele.(*Client)