Zhuang's Diary

c := make(chan bool)    //创建一个无缓冲的bool型Channel
c <- x                  //向一个Channel发送一个值
<- c                    //从一个Channel中接收一个值
x = <- c                //从Channel c接收一个值并将其存储到x中
x, ok = <- c            //从Channel接收一个值，如果channel关闭了或没有数据，那么ok将被置为false

func (m *SomeController) PorcessSomeTask() {
    var task models.Task
    if err := task.Parse(m.Ctx.Request); err != nil {
        m.Data["json"] = err 
        m.ServeJson()
        return
    }
    go task.Process()
    m.ServeJson()

var TASK_CHANNEL = make(chan models.Task)

// 然后，启动多个goroutine：
for i := 0; i < WORKER_NUM; i ++ {
    go func() {
        for {
            select {
            case task := <- TASK_CHANNEL:
                task.Process()
            }
        }
    } ()
}

// 服务端接收到请求之后，将任务传入channel中即可：
func (m *SomeController) PorcessSomeTask() {
    var task models.Task
    if err := task.Parse(m.Ctx.Request); err != nil {
        m.Data["json"] = err 
        m.ServeJson()
        return
    }
    // go task.Process()
    TASK_CHANNEL <- task
    m.ServeJson()
}

var TASK_CHANNEL = make(chan models.Task, TASK_CHANNEL_LEN)

select {
case TASK_CHANNEL <- task:
    //do nothing
default:
    //warnning!
    return fmt.Errorf("TASK_CHANNEL is full!")
}
//...

type TaskResponse struct {
    //...
}

type Task struct {
    TaskParameter   SomeStruct
    ResChan         *chan TaskResponse
}

//...

task := Task {
    TaskParameter   : xxx,
    ResChan         : make(chan TaskResponse),
}

TASK_CHANNEL <- task
res := <- task.ResChan
//...

var wg sync.WaitGroup
for i := 0; i < someLen; i ++ {
    wg.Add(1)
    go func(t Task) {
        defer wg.Done()
        //对某一段子任务进行处理
    } (tasks[i])
}

wg.Wait()
//处理剩下的工作

task := Task {
    TaskParameter   : xxx,
    ResChan         : make(chan TaskResponse),
}

select {
case res := <- task.ResChan:
    //...
case <- time.After(PROCESS_MAX_TIME):
    //处理超时
}

// 执行方
func doTask(done <-chan struct{}, tasks <-chan Task) (chan Result) {
    out := make(chan Result)
    go func() {
        // close 是为了让调用方的range能够正常退出
        defer close(out)
        for t := range tasks {
            select {
            case result <-f(t):
                // do task
            case <-done:
                return
            }
        }
    }()

    return out
}

// 调用方
func Process(tasks <-chan Task, num int) {
    done := make(chan struct{})
    out := doTask(done, tasks)

    go func() {
        <- time.After(MAX_TIME)
        //done <-struct{}{}

        //通知所有的执行goroutine退出
        close(done)
    }()

    // 因为goroutine执行完毕，或者超时，导致out被close，range退出
    for res := range out {
        fmt.Println(res)
        //...
    }
}

syntax = "proto3";
package example;
service FormatData {
    rpc DoFormat(Data) returns (Data){} 
    // 此处定义的方法为简单RPC。其余的服务器端流式 RPC，客户端流式 RPC，双向流式 RPC请查询参考链接
}
message Data {
    string text = 1;
}

package main
import (
	"gRPCDemo/example"
	"net"
	"google.golang.org/grpc"
	"google.golang.org/grpc/reflection"
	"golang.org/x/net/context"
	"strings"
	"log"
)
// 定义监听地址
const (
	HOST string = "localhost"
	PORT string = "8080"
)
// 定义接口
type FormatData struct{}
func (fd *FormatData) DoFormat(ctx context.Context, in *example.Data) (out *example.Data, err error) {
	str := in.Text
	out = &example.Data{Text: strings.ToUpper(str)}
	return out, nil
}
// 直接在 main 方法中注册接口
func main() {
	// 启动服务器
	// 指定我们期望客户端请求的监听端口
	listener, err := net.Listen("tcp", HOST+":"+PORT)
	if err != nil {
		log.Fatalln("faile listen at: " + HOST + ":" + PORT)
	} else {
		log.Println("Demo server is listening at: " + HOST + ":" + PORT)
	}
	// 创建 gRPC 服务器的一个实例
	rpcServer := grpc.NewServer()
	// 在 gRPC 服务器注册我们的服务实现
	example.RegisterFormatDataServer(rpcServer, &FormatData{})
	reflection.Register(rpcServer)
	// 实现阻塞等待，直到进程被杀死或者 Stop() 被调用
	if err = rpcServer.Serve(listener); err != nil {
		log.Fatalln("faile serve at: " + HOST + ":" + PORT)
	}
}

package main
import (
	"google.golang.org/grpc"
	"log"
	"gRPCDemo/example"
	"golang.org/x/net/context"
)
// 定义请求地址
const (
	ADDRESS string = "localhost:8080"
)
// main 方法实现对 gRPC 接口的请求
func main() {
	// 为了调用服务方法，我们首先创建一个 gRPC channel 和服务器交互。
	// 可以使用 DialOptions 在 grpc.Dial 中设置授权认证（如， TLS，GCE认证，JWT认证）
	conn, err := grpc.Dial(ADDRESS, grpc.WithInsecure())
	if err != nil {
		log.Fatalln("Can't connect: " + ADDRESS)
	}
	defer conn.Close()
	// 一旦 gRPC channel 建立起来，我们需要一个客户端 存根 去执行 RPC
	client := example.NewFormatDataClient(conn)
	// 调用简单 RPC
	resp,err := client.DoFormat(context.Background(), &example.Data{Text:"hello,world!"})
	// 如果调用没有返回错误，那么我们就可以从服务器返回的第一个返回值中读到响应信息
	if err != nil {
		log.Fatalln("Do Format error:" + err.Error())
	}
	log.Println(resp.Text)
}

package main

import (
	"github.com/golang/protobuf/proto"
	"protocDemo/example"
	"log"
)

func main() {
	// 为 AllPerson 填充数据
	p1 := example.Person{
		Id:*proto.Int32(1),
		Name:*proto.String("xieyanke"),
	}

	p2 := example.Person{
		Id:2,
		Name:"gopher",
	}

	all_p := example.AllPerson{
		Per:[]*example.Person{&p1, &p2},
	}

	// 对数据进行序列化
	data, err := proto.Marshal(&all_p)
	if err != nil {
		log.Fatalln("Mashal data error:", err)
	}
	println(data)

	// 对已经序列化的数据进行反序列化
	var target example.AllPerson
	err = proto.Unmarshal(data, &target)
	if err != nil{
		log.Fatalln("UnMashal data error:", err)
	}

	println(target.Per[0].Name) // 打印第一个 person Name 的值进行反序列化验证
}