What you are wasting today is tomorrow for those who died yesterday; what you hate now is the future you can not go back.

你所浪费的今天是昨天死去的人奢望的明天； 你所厌恶的现在是未来的你回不去的曾经。 

 如何通过简单的tcp / ip连接将数据从进程a发送到进程b？

       在许多情况下，使用更高级别的网络协议无疑会做得更好，从而将所有技术细节隐藏在一个奇特的API下面。并且已经有很多可供选择的，取决于需要：消息队列协议，grpc，protobuf，flatbuffers，restful web api，websockets等等。

 

       但是，在某些情况下（特别是在小型项目中），您选择的任何方法可能看起来完全过大。

1.  connections是一个io流

net.Conn实现了 io.Reader, io.Writer, io.Closer接口。 所以我们这是像使用io流一样来使用TCP 链接。

首先我们来看看 Golang源码的 io 包中的这三个类型的定义：

type Reader interface {
	Read(p []byte) (n int, err error)
}

type Writer interface {
	Write(p []byte) (n int, err error)
}

type Closer interface {
	Close() error
}

再来看看Golang源码中net包Conn 类型的定义：

type Conn interface {
	
	Read(b []byte) (n int, err error)


	Write(b []byte) (n int, err error)

	
	Close() error

	LocalAddr() Addr

	RemoteAddr() Addr

	SetDeadline(t time.Time) error

	
	SetReadDeadline(t time.Time) error

	
	SetWriteDeadline(t time.Time) error
}

那么我们可以通过TCP链接发送string字符串了， 但是如何发送复杂的类型呢？

2. Go 编码复杂类型

当涉及通过网络发送结构化数据时，很容易想到json,  但是Go自身提供了一个gob包直接在io流数据上操作，序列化和反序列化数据，不需要json那样添加标签， 然后再费力的json.Unmarshal()转为二进制数据.

 

3. 通过TCP发送字符串数据的基本要素：

    1.发送方

            1. 打开一个接收进程的链接

            2. 写入字符串

            3. 关闭链接

        Golang的net包已经提供了以上的所有方法。

 

    ResolveTCPAddr() 接受一个表示TCP地址的字符串（localhost, 127.0.0.1:80,  [::1]:80 都表示本地80端口）， 返回一个net.TCPAddr() ,如果无法解析此地址将返回错误。

    DialTCP()接受一个net.Addr()然后连接到此地址，成功后返回一个打开的net.TCPConn链接对象。

    如果我们不需要对拨号设置更为详细。我们可以直接使用net.Dial 来代替。

    如果链接成功， 将可以将链接对象封装为一个bufio.ReadWriter， 

type ReadWriter struct {
    *Reader
    *Writer
}

我们就可以使用 ReadString() Writestring() ReadBytes()第方法读取数据

注意的是，缓冲写入需要在写入后调用flush（），以便将所有数据转发到底层网络连接

    2. 发送方

        1. 开始监听本地的端口

        2. 当一个接受到请求后， 发起一个goroutine 来处理此请求

        3. 在这个goroutine里读取数据 ， 可选的发送响应。

        4. 关闭链接

4. 复杂类型的处理

    服务端根据请求的数据类型，给出对象的处理方式。简要的运行方式：

    第一步： 当listen()接受到一个新链接时， 生成一个新的goroutine来执行对应数据类型的请求方法HandleMessage().该函数从连接读取命令名称，从映射中查找适当的处理函数，并调用该函数。

    第二部： 选定的处理函数读取并处理请求数据。

    

   详细描述：

        发起请求-> 服务端监听-> 判断请求数据类型以及对应的处理方法-> 新goroutine -> 具体的处理方法处理。

 

详细的代码：

    1. 项目目录架构

    

    2. 库文件

        

package lib

import (
	"bufio"
	"net"
	"github.com/pkg/errors"
	"fmt"
	"sync"
	"io"
	"strings"
	"encoding/gob"
)

// 混合类型的struct
type ComplexData struct{
	 N int
	 S  string
	 M map[string]int
	 P []byte
	 C *ComplexData
}

const(
	Port = ":61000" // 服务端接受的端口
)

/**
	net.Conn 实现了io.Reader  io.Writer  io.Closer接口
	Open 返回一个有超时的TCP链接缓冲readwrite
 */
func Open(addr string) (*bufio.ReadWriter, error) {
	// Dial the remote process.
	// Note that the local port is chosen on the fly. If the local port
	// must be a specific one, use DialTCP() instead.
	fmt.Println("Dial " + addr)
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		return nil, errors.Wrap(err, "Dialing "+addr+" failed")
	}
	return bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)), nil
}

 type HandleFunc func(*bufio.ReadWriter)

 type EndPoint struct{
 	listener net.Listener
	 // handlefunc是一个处理传入命令的函数类型。 它接收打包在一个读写器界面中的开放连接。
	 handler map[string]HandleFunc

 	// map不是线程安全的，所以需要读写锁控制
 	m sync.RWMutex
 }

func NewEndPoint() *EndPoint{
	return &EndPoint{
		handler:map[string]HandleFunc{},
	}
}
// 添加数据类型处理方法
func (e *EndPoint)AddHandleFunc(name string , f HandleFunc){
	e.m.Lock()
	e.handler[name] = f
	e.m.Unlock()
}

// 验证请求数据类型，并发送到对应处理函数
func (e *EndPoint)handleMessage(conn net.Conn){
	rw := bufio.NewReadWriter(bufio.NewReader(conn),
		bufio.NewWriter(conn))
	defer conn.Close()
	for{
		cmd, err := rw.ReadString('\n')
		switch  {
		case err == io.EOF:
			fmt.Println("读取完成.")
			return
		case err != nil:
			fmt.Println("读取出错")
			return
		}

		cmd = strings.Trim(cmd, "\n ")
		e.m.RLock()
		handleCmd , ok := e.handler[cmd]
		if !ok{
			fmt.Println("未注册的请求数据类型.")
			return
		}
		//具体处理链接数据
		handleCmd(rw)
	}
}

func (e *EndPoint) Listen()error{
	var err error
	e.listener, err = net.Listen("tcp", Port)
	if err != nil{
		return errors.Wrap(err , "TCP服务无法监听在端口"+Port)
	}
	fmt.Println(" 服务监听成功：",e.listener.Addr().String())
	for{
		conn, err := e.listener.Accept()
		if err != nil{
			fmt.Println("心请求监听失败!")
			continue
		}
		// 开始处理新链接数据
		go e.handleMessage(conn)
	}

}

func HandleStrings(rw *bufio.ReadWriter){
	s, err := rw.ReadString('\n')
	if err!= nil{
		fmt.Println("链接无法读取.")
		return
	}

	s = strings.Trim(s , "\n ")
	// ....
	_, err = rw.WriteString("处理完成......\n")
	if err != nil{
		fmt.Println("链接写入响应失败")
		return
	}
	// 写入底层网络链接
	err = rw.Flush()
	if err != nil{
		fmt.Println("Flush写入失败")
		return
	}
}

func HandleGob(rw *bufio.ReadWriter){
	var data ComplexData

	dec := gob.NewDecoder(rw)
	err := dec.Decode(&data)
	if err != nil{
		fmt.Println("无法解析的二进制数据.")
		return
	}
	fmt.Println("输出：", data, data.C)
}

3. 服务文件

server.go

package main

import(
	. "tcpNetWorking/lib"
	"fmt"
	"github.com/pkg/errors"
)

func server()error{
	endpoint := NewEndPoint()

	endpoint.AddHandleFunc("string", HandleStrings)
	endpoint.AddHandleFunc("gob", HandleGob)

	// 开始监听
	return endpoint.Listen()
}

func main(){
	err := server()
	if err != nil {
		fmt.Println("Error:", errors.WithStack(err))
	}
}

 

client.go

package main

import (
	"fmt"
	. "tcpNetWorking/lib"
	"github.com/pkg/errors"
	"encoding/gob"
	"strconv"
	"log"
)

func client(ip string) error {
	cpData := ComplexData{
		N: 10,
		S: "测试string 数据",
		M: map[string]int{"A": 1, "B": 2},
		P: []byte("测试[]byte数据"),
		C: &ComplexData{
			N: 256,
			S: "Recursive structs? Piece of cake!",
			M: map[string]int{"01": 1, "10": 2, "11": 3},
		},
	}
	rw, err := Open(ip + Port)
	if err != nil {
		fmt.Println("客户端无法链接改地址：" + ip + Port)
		return err
	}
	n, err := rw.WriteString("string\n")
	if err != nil {
		return errors.Wrap(err, "Could not send the STRING request ("+strconv.Itoa(n)+" bytes written)")
	}
	n, err = rw.WriteString("Additional data.\n")
	if err != nil {
		return errors.Wrap(err, "Could not send additional STRING data ("+strconv.Itoa(n)+" bytes written)")
	}
	err = rw.Flush()
	if err != nil {
		return errors.Wrap(err, "Flush failed.")
	}

	// Read the reply.
	response, err := rw.ReadString('\n')
	if err != nil {
		return errors.Wrap(err, "Client: Failed to read the reply: '"+response+"'")
	}

	log.Println("STRING request: got a response:", response)

	log.Println("Send a struct as GOB:")
	log.Printf("Outer complexData struct: \n%#v\n", cpData)
	log.Printf("Inner complexData struct: \n%#v\n", cpData.C)
	enc := gob.NewEncoder(rw)
	n, err = rw.WriteString("gob\n")
	if err != nil {
		return errors.Wrap(err, "Could not write GOB data ("+strconv.Itoa(n)+" bytes written)")
	}
	err = enc.Encode(cpData)
	if err != nil {
		return errors.Wrapf(err, "Encode failed for struct: %#v", cpData)
	}
	err = rw.Flush()
	if err != nil {
		return errors.Wrap(err, "Flush failed.")
	}
	return nil
}

func main(){
	err := client("localhost")
	if err != nil {
		fmt.Println("Error:", errors.WithStack(err))
	}
}

 

 

逻辑基本上与我之前写的web 路由服务差不多， 只是数据处理使用gob包 二进制形式。 看一顺带看一下

https://my.oschina.net/90design/blog/1604539

 

 

---