第一次,站长亲自招 Gopher 了>>>

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

---

本文来自:开源中国博客

感谢作者:90design

查看原文:Golang 通过tcp / ip发送数据

11743 次点击

上一篇:golang 几种字符串的连接方式

下一篇:Go 指南

net io 函数 处理函数

• 编辑
• 预览