自增人生

/**
引用自 golang.org/src/log/log.go
const (
Ldate         = 1 << iota     // 日期: 2009/01/23
Ltime                         // 时间: 01:23:23
Lmicroseconds                 // 毫秒级时间: 01:23:23.123123.  覆盖 Ltime.
Llongfile                     // 完整的源码文件路径及行号: /a/b/c/d.go:23
Lshortfile                    // 短路径及行号: d.go:23. 会覆盖 Llongfile
LUTC                          // 如果设置了Ldata或Ltime，采用UTC取代本地时区
LstdFlags     = Ldate | Ltime // 标准日志初始值
)
*/

// 初始化后，所有直接调用log包的日志输出都会受影响
func init() {
  // 设置日志的前缀信息
  log.SetPrefix("[LogPrefix] ")
  // 设置日志的中段标示，参考上边的注释
  log.SetFlags(log.Ldate | log.Lmicroseconds | log.Lshortfile)
}

package main

import "log"

func main() {
  log.Println("hello world")
}

// ------------以下是程序输出--------------
[LogPrefix] 2019/03/09 14:34:06.763601 main.go:6: hello world

var (
  Trace   *log.Logger // 普通跟踪调试信息
  Info    *log.Logger // 特殊信息
  Warning *log.Logger // 警告日志
  Error   *log.Logger // 错误日志(输出到文件)
)

func init() {
  file, err := os.OpenFile("error.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
  if err != nil {
    log.Fatalln("Failed to open error file: ", err)
  }

// 因为ioutil.Discard，所有通过Trace打印的日志都不会输出
    Trace = log.New(ioutil.Discard, "TRACE: ", log.Ldate|log.Ltime|log.Lshortfile)
    Info = log.New(os.Stdout, "INFO: ", log.Ldate|log.Ltime|log.Lshortfile)
    Warning = log.New(os.Stdout, "WARNING: ", log.Ldate|log.Ltime|log.Lshortfile)
    // io.MultiWriter表示多种输出渠道，即同时打印到屏幕和文件
    Error = log.New(io.MultiWriter(file, os.Stderr), "ERROR: ", log.Ldate|log.Ltime|log.Llongfile)
}

// User Github用户账户信息
// 每个变量最后用反引号标示的字符串是——标签
// 标签可为之后的json.Decoder提供映射依据
type User struct {
  Login             string `json:"login"`
  ID                int64  `json:"id"`
  AvatarURL         string `json:"avatar_url"`
  GravatarID        string `json:"gravatar_id"`
  URL               string `json:"url"`
  HTMLURL           string `json:"html_url"`
  FollowersURL      string `json:"followers_url"`
  FollowingURL      string `json:"following_url"`
  GistsURL          string `json:"gists_url"`
  StarredURL        string `json:"starred_url"`
  SubscriptionsURL  string `json:"subscriptions_url"`
  OrganizationsURL  string `json:"organizations_url"`
  ReposURL          string `json:"repos_url"`
  EventsURL         string `json:"events_url"`
  RecievedEventsURL string `json:"recieved_events_url"`
  Type              string `json:"type"`
  SiteAdmin         bool   `json:"site_admin"`
  Name              string `json:"name"`
  Company           string `json:"company"`
  Blog              string `json:"blog"`
  Location          string `json:"location"`
  Email             string `json:"email"`
  Hireable          string `json:"hireable"`
  Bio               string `json:"bio"`
  PublicRepos       int32  `json:"public_repos"`
  PublicGists       int32  `json:"public_gists"`
  Followers         int32  `json:"followers"`
  Following         int32  `json:"following"`
  CreateAt          string `json:"create_at"`
  UpdateAt          string `json:"update_at"`
}

// DeserializeToType 获取用户信息，并反序列化为User类型
func DeserializeToType(name string) {
  // 根据用户名从GitHub获取对应用户的json信息
  resp, err := http.Get("https://api.github.com/users/" + name)
  if err != nil {
  log.Println("ERROR: ", err)
  }

  defer resp.Body.Close()

  var user User
  // 通过Decode将响应内容反序列化为对象
  err = json.NewDecoder(resp.Body).Decode(&user)
  if err != nil {
  log.Println("ERROR: ", err)
  }

  fmt.Printf("user: %v\n", user)
}

func DeserializeToMap(name string) {
  resp, err := http.Get("https://api.github.com/users/" + name)
  if err != nil {
    log.Fatalln("ERROR: ", err)
  }

  defer resp.Body.Close()

  // 读取响应Body并转化为[]byte数据结构
  content, err := ioutil.ReadAll(resp.Body)
  if err != nil {
    log.Fatalln("ERROR: ", err)
  }

  var user map[string]interface{}
  // 通过Unmarshal将[]byte转换为字典
  err = json.Unmarshal(content, &user)
  if err != nil {
    log.Fatalln("ERROR: ", err)
  }

  fmt.Println("Username: ", user["login"])
  fmt.Println("Followers: ", user["followers"])
}

func SerializeType() {
  user := &User{
    Login: "张三",
    ID:    9527,
    URL:   "https://api.github.com/users/ZhangSan",
  }

  // 带缩进格式的序列化，缩进为4个空格
  data, err := json.MarshalIndent(user, "", "    ")
  if err != nil {
    log.Fatalln("ERROR: ", err)
  }

  fmt.Println(string(data))
}

func SerializeMap() {
  c := make(map[string]interface{})
  c["name"] = "张三"
  c["id"] = 9527

  // 不带缩进格式化的序列化
  data, err := json.Marshal(c)
  if err != nil {
    log.Fatalln("ERROR: ", err)
  }

  fmt.Println(string(data))
}

func main() { 
  // 创建一个Buffer值，并将一个字符串写入Buffer 
  // 使用实现io.Writer的Write方法 
  var b bytes.Buffer
  b.Write([]byte("Hello "))

  // 使用Fprintf来将一个字符串拼接到Buffer里 
  // 将bytes.Buffer的地址作为io.Writer类型值传入 
  fmt.Fprintf(&b, "World!")

  // 将Buffer的内容输出到标准输出设备 
  // 将os.File值的地址作为io.Writer类型值传入 
  b.WriteTo(os.Stdout)
} 

func testRunner() {
  log.Println("Runner test starting work...")

  // 新建一个runner，并强制每个并发任务的超时时间为5秒
  r := runner.New(3 * time.Second)
  // 循环创建10个并发任务，并将其丢给runner管理
  for i := 0; i < 5; i++ {
    r.Add(func(id int) {
      // 这里只是模拟，每个并发任务都是睡眠它自身id的秒数
      log.Printf("Processor - Task #%d.\n", id)
      time.Sleep(time.Duration(id) * time.Second)
      log.Printf("Task #%d done.\n", id)
    })
  }

  // 一次性启动runner内部的全部并发任务
  if err := r.Start(); err != nil {
    switch err {
    case runner.ErrTimeout:
      // 当并发任务中有任务执行超时，就立即返回
      log.Println("Terminating due to timeout.")
      os.Exit(1)
    case runner.ErrInterrupt:
      // 当程序被ctrl+c时，强制结束所有并发任务
      log.Println("Terminating due to interrupt.")
      os.Exit(2)
    }
  }

  log.Println("Process end.")
}

//-------------程序输出---------------
2019/03/03 14:48:41 Runner test starting work...
2019/03/03 14:48:41 Processor - Task #2.
2019/03/03 14:48:41 Processor - Task #4.
2019/03/03 14:48:41 Processor - Task #3.
2019/03/03 14:48:41 Processor - Task #0.
2019/03/03 14:48:41 Task #0 done.
2019/03/03 14:48:41 Processor - Task #1.
2019/03/03 14:48:42 Task #1 done.
2019/03/03 14:48:43 Task #2 done.
// ----第3个及以后的任务因为要睡3秒以上，肯定会超时----
2019/03/03 14:48:44 Terminating due to timeout.
// ----如何运行过程中按ctrl+c，会安全退出并提示----
^C2019/03/03 14:49:36 Terminating due to interrupt.

// Runner 在指定的超时时间内完成一组任务
// 并且在这个时间周期内接收系统的中断信号来结束这组任务
type Runner struct {
  // 从系统接收中断信号的通道
  interrupt chan os.Signal
  // 任务已完成的报告通道
  complete chan error
  // 任务超时的报告通道
  timeout <-chan time.Time
  // 任务列表
  tasks []func(id int)
}

// ErrTimeout 任务执行超时时返回
var ErrTimeout = errors.New("received timeout")

// ErrInterrupt 收到系统中断信号时返回
var ErrInterrupt = errors.New("received interrupt")

// New 创建Runner的工厂函数
func New(d time.Duration) *Runner {
  return &Runner{
    interrupt: make(chan os.Signal, 1),
    complete:  make(chan error),
    timeout:   time.After(d),
  }
}

// Add Runner的方法，将多个任务添加到Runner的任务列表中
func (r *Runner) Add(tasks ...func(int)) {
  r.tasks = append(r.tasks, tasks...)
}

// Start Runner的方法，启动所有任务，并监听通道事件
func (r *Runner) Start() error {
  // 开始接收系统的中断通知
  signal.Notify(r.interrupt, os.Interrupt)

  // 通过gorouting并行启动所有任务列表中的任务
  var wg sync.WaitGroup
  wg.Add(len(r.tasks))
  for i, t := range r.tasks {
    go func(id int, task func(int)) {
      task(id)
      wg.Done()
    }(i, t)
  }
  // 等待所有任务执行完成，并给“已完成通道”一个报告
  go func() {
    wg.Wait()
    r.complete <- nil
  }()

  // 获取无缓冲通道数据时，如果没准备好，会被阻塞
  select {
  case err := <-r.complete: // 任务正常实行完返回任务自身的“错误标示”
    return err
  case <-r.timeout: // 任务执行超时，返回超时错误
    return ErrTimeout
  case <-r.interrupt: // 如果收到ctrl+C则停止接收后续的信号
    signal.Stop(r.interrupt)
    return ErrInterrupt
  }
}

const (
  maxGoroutines  = 25 // 要使用的goroutine的数量
  pooledResource = 5  // 池中的资源的数量
)

// dbConnection 模拟要共享的资源
type dbConnection struct {
  ID int32
}

// Close 实现io.Closer.Close接口，释放资源
// 让其可以被Pool管理
func (dbConn *dbConnection) Close() error {
  log.Println("Close: Connection", dbConn.ID)
  return nil
}

// idCounter 用来给每个连接分配唯一id
var idCounter int32

// createConnection 创建唯一id的连接
func createConnection() (io.Closer, error) {
  id := atomic.AddInt32(&idCounter, 1)
  log.Println("Create: New Connection", id)

  return &dbConnection{id}, nil
}

// testPool Pool测试用例
func testPool() {
  var wg sync.WaitGroup
  wg.Add(maxGoroutines)

  // 创建管理连接池，并创建N个“连接”资源，加入池中
  p, err := pool.New(createConnection, pooledResource)
  if err != nil {
    log.Println(err)
  }

  // 创建M个并发任务，模拟查询数据库
  for query := 0; query < maxGoroutines; query++ {
    go func(q int) {
      performQueries(q, p)
      wg.Done()
    }(query)
  }

  wg.Wait()
  log.Println("Shutdown Program")
  p.Close() // 关闭池
}

func performQueries(query int, p *pool.Pool) {
  conn, err := p.Acquire()
  if err != nil {
    log.Println(err)
    return
  }

  // 完成查询后，将资源释放会池里
  defer p.Release(conn)

  // 用随机睡眠1000微妙内的时长，来模拟查询中的耗时
  time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
  log.Printf("QID[%d] CID[%d]\n", query, conn.(*dbConnection).ID)
}

// Pool 资源池
// 管理一组资源，可以安全地在多个goroutine共享
// 实现 io.Closer接口
type Pool struct {
  m        sync.Mutex
  resource chan io.Closer
  factory  func() (io.Closer, error)
  closed   bool
}

// ErrPoolClosed 资源池已关闭的错误标示
var ErrPoolClosed = errors.New("Pool has been closed")

// New 创建Pool的工厂函数
func New(fn func() (io.Closer, error), size uint) (*Pool, error) {
  if size <= 0 {
    return nil, errors.New("size value too small")
  }

  return &Pool{
    factory:  fn,
    resource: make(chan io.Closer, size),
  }, nil
}

// Acquire 从资源池中获取一个资源
func (p *Pool) Acquire() (io.Closer, error) {
  select {
  case r, ok := <-p.resource:
    // 检查是否有空闲资源
    log.Println("Acquire:", "Shared Resource")
    if !ok {
      return nil, ErrPoolClosed
    }
    return r, nil
  default:
    // 如果没有可用资源，就创建一个
    log.Println("Acquire:", "New Resource")
    return p.factory()
  }
}

// Release 释放一个资源，将其放回资源池
func (p *Pool) Release(r io.Closer) {
  p.m.Lock()
  defer p.m.Unlock()

  // 如果该资源已经被关闭，销毁这个资源
  if p.closed {
    r.Close()
    return
  }

  select {
  case p.resource <- r:
    // 试图将该资源加入队列
    log.Println("Release:", "In Queue")
  default:
    // 如果队列已满，关闭这个资源
    log.Println("Release:", "Closing")
    r.Close()
  }
}

// Close 关闭资源池中的所有资源，并停止工作
func (p *Pool) Close() {
  p.m.Lock()
  defer p.m.Unlock()

  if p.closed {
    return
  }

  p.closed = true

  close(p.resource)

  // 关闭所有资源
  for r := range p.resource {
    r.Close()
  }
}

var names = []string{
  "steve",
  "bob",
  "mary",
  "therese",
  "jason",
}

type namePrinter struct {
  name string
}

// Task 实现Worker接口
func (m *namePrinter) Task() {
  log.Println(m.name)
  time.Sleep(time.Second)
}

func testWork() {
  // 设置工作池的“工位”为2，即每次只能有两个人工作
  p := work.New(2)

  var wg sync.WaitGroup
  wg.Add(100 * len(names)) // 每人肩负100项任务，共5人

  // 一次性把5 * 100个任务全部丢到工作池中
  // 相当于创建了500个goroutine
  for i := 0; i < 100; i++ {
    for _, name := range names {
      np := namePrinter{name: name}
      go func() {
        p.Run(&np) // 将对象的任务丢到工作池中统一管理执行
        wg.Done()
      }()
    }
  }

  // 等待所有任务在工作池中被完成
  wg.Wait()
  p.Shutdown()
}

// Worker 必须满足接口，才能使用工作池
type Worker interface {
  Task()
}

// Pool 工作池，相当于goroutines池管理
type Pool struct {
  work chan Worker
  wg   sync.WaitGroup
}

// New 创建工作池的工厂函数
func New(maxGoroutines int) *Pool {
  p := Pool{
    work: make(chan Worker),
  }

  p.wg.Add(maxGoroutines)
  for i := 0; i < maxGoroutines; i++ {
    go func() {
      // p.work是通道，所有创建goroutine之后
      // for循环会被阻塞，直到p.work被关闭为止
      for w := range p.work {
        w.Task()
      }
      p.wg.Done()
    }()
  }

  return &p
}

// Run 提交工作到工作池
func (p *Pool) Run(w Worker) {
  p.work <- w
}

// Shutdown 等待所有goroutines结束
func (p *Pool) Shutdown() {
  close(p.work)
  p.wg.Wait()
}

func main() {
  var wg sync.WaitGroup
  wg.Add(10) // WaitGroup计数加10

  for i := 0; i < 10; i++ {
    // 创建goroutine
    go func(id int) {
      fmt.Printf("Goroutine-%d\n", id)
      wg.Done() // WaitGroup计数减1
    }(i)
  }

  wg.Wait() // 等待，直到WaitGroup计数为0
}

var count int // 声明一个全局变量

func main() {
  var wg sync.WaitGroup
  wg.Add(5)

  for i := 0; i < 5; i++ {
    go func() {
      defer wg.Done() // 👈匿名函数退出时调用

      var num = count // 读全局变量
      fmt.Printf("num = %d\n", num)
      count = num + 1 // 写全局变量
    }()
  }

  wg.Wait()
  // 所有goroutine结束后，count值并不为5
  fmt.Printf("count = %d\n", count)
}
// ----------程序输出结果(每次都是随机的)--------
num = 0
num = 1
num = 0
num = 0
num = 0
count = 1

var wg sync.WaitGroup

// 接收端goroutine
func receive(data chan int) {
  num := <-data // 将通道数据读取到变量
  fmt.Printf("received number: %d\n", num)
  wg.Done()
}

// 发送端goroutine
func send(data chan int, num int) {
  fmt.Printf("sent number: %d\n", num)
  data <- num // 将数据写入通道
  wg.Done()
}

// UseChannel 通道的基本使用
func UseChannel() {
  channel := make(chan int) // 创建一个通道
  wg.Add(2)
  go receive(channel)
  go send(channel, 20)
  wg.Wait()
  close(channel) // 关闭通道
}

func BufferChannel() {
  // 创建一个长度为10的有缓冲通道
  bufferedChannel := make(chan int, 10)
  wg.Add(10)

  // 启动5个goroutine接收数字
  for i := 0; i < 5; i++ {
    go receive(bufferedChannel)
  }
  // 启动5个goroutine发送随机5个数字
  for i := 0; i < 5; i++ {
    go send(bufferedChannel, rand.Intn(100))
  }

  wg.Wait()
  close(bufferedChannel)
}

//---------------程序输出结果------------
sent number: 81
received number: 81
sent number: 81
sent number: 87
sent number: 47
received number: 81
received number: 87
received number: 47
sent number: 59
received number: 59