fastnet

README

fastnet 简介

• fastnet 是一个高性能、轻量级、非阻塞的 TCP 网络库，用户可以使用 fastnet 来实现自己的应用层网络协议，从而构建出自己的应用层网络应用；
• 在某些极端的网络业务场景，比如海量连接、高频短连接、网络小包等场景，fastnet 在性能和资源占用上都超过 Go 原生 net 包的goroutine-per-connection 模式。

fastnet 特性

• 使用 Epoll 水平触发的IO多路复用技术，非阻塞IO，使用 Reactor 模式；
• 使用多线程充分利用多核CPU，使用动态扩容 Ring Buffer 实现读写缓冲区；
• 支持异步读写操作、支持 SO_REUSEPORT 端口重用；
• 灵活的事件定时器，可以定时任务，延时任务；
• 支持 WebSocket，同时支持自定义协议，处理 TCP 粘包；

TODO：

• 更多的负载方式（轮询（Round-Robin）、一致性哈希（consistent hashing）等）；

Installation

1. 获得并安装 fastnet

$ go get -u github.com/dongxiem/fastnet

1. 进行 import

import "github.com/dongxiem/fastnet"

fastnet 性能测试

• bench-echo
• bench-pingpong

Example

package main

import (
	"flag"
	"net/http"
	_ "net/http/pprof"
	"strconv"
	"time"

	"github.com/Dongxiem/fastnet"
	"github.com/Dongxiem/fastnet/connection"
	"github.com/Dongxiem/fastnet/log"
	"github.com/Dongxiem/fastnet/tool/sync/atomic"
)

// 定义 example 类型，并实现 Handler 接口对应的所有方法
type example struct {
	Count atomic.Int64
}

func (s *example) OnConnect(c *connection.Connection) {
	s.Count.Add(1)
	//log.Println(" OnConnect ： ", c.PeerAddr())
}
func (s *example) OnMessage(c *connection.Connection, ctx interface{}, data []byte) (out []byte) {
	//log.Println("OnMessage")
	out = data
	return
}

func (s *example) OnClose(c *connection.Connection) {
	s.Count.Add(-1)
	//log.Println("OnClose")
}

func main() {
	go func() {
		if err := http.ListenAndServe(":6060", nil); err != nil {
			panic(err)
		}
	}()

	handler := new(example)
	var port int
	var loops int

	flag.IntVar(&port, "port", 1833, "server port")
	flag.IntVar(&loops, "loops", -1, "num loops")
	flag.Parse()

	// new 一个 server，根据传递进来的 loops 进行工作线程循环的调配
	s, err := fastnet.NewServer(handler,
		fastnet.Network("tcp"),
		fastnet.Address(":"+strconv.Itoa(port)),
		fastnet.NumLoops(loops))
	if err != nil {
		panic(err)
	}

	// 每两秒打印一下 handle 统计
	s.RunEvery(time.Second*2, func() {
		log.Info("connections :", handler.Count.Get())
	})
	// 启动 server
	s.Start()
}

Documentation

Index

• type Handler
• type Option
  • func Address(a string) Option
  • func IdleTime(t time.Duration) Option
  • func Network(n string) Option
  • func NumLoops(n int) Option
  • func Protocol(p connection.Protocol) Option
  • func ReusePort(reusePort bool) Option
• type Options
• type Server
  • func NewServer(handler Handler, opts ...Option) (server *Server, err error)
  • func (s *Server) Options() Options
  • func (s *Server) RunAfter(d time.Duration, f func()) *timingwheel.Timer
  • func (s *Server) RunEvery(d time.Duration, f func()) *timingwheel.Timer
  • func (s *Server) Start()
  • func (s *Server) Stop()

Examples

• Server.RunAfter
• Server.RunEvery

Types

type Handler

type Handler interface {
	connection.CallBack
	OnConnect(c *connection.Connection)
}

Handler：Server 注册接口

type Option

type Option func(*Options)

Option ...

func Address

func Address(a string) Option

Address：server 监听地址

func IdleTime

func IdleTime(t time.Duration) Option

IdleTime：最大空闲时间（秒）

func Network

func Network(n string) Option

Network：暂时只支持tcp

func NumLoops

func NumLoops(n int) Option

NumLoops：work eventloop 的数量

func Protocol

func Protocol(p connection.Protocol) Option

Protocol：数据包处理

func ReusePort

func ReusePort(reusePort bool) Option

ReusePort：设置 SO_REUSEPORT

type Options

type Options struct {
	Network   string // 网络协议
	Address   string // 监听端口地址
	NumLoops  int    // work 协程个数，负责处理已连接客户端的读写事件
	ReusePort bool   // 是否开启端口复用

	IdleTime time.Duration       // 最大空闲时间（秒）
	Protocol connection.Protocol // 连接协议
	// contains filtered or unexported fields
}

Options：服务配置

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server：fastnet Server

func NewServer

func NewServer(handler Handler, opts ...Option) (server *Server, err error)

NewServer：创建 Server

func (*Server) Options

func (s *Server) Options() Options

Options：返回 options

func (*Server) RunAfter

func (s *Server) RunAfter(d time.Duration, f func()) *timingwheel.Timer

RunAfter：延时任务开启

Example

handler := new(example)

s, err := NewServer(handler,
	Network("tcp"),
	Address(":1833"),
	NumLoops(8),
	ReusePort(true))
if err != nil {
	panic(err)
}

go s.Start()
defer s.Stop()

s.RunAfter(time.Second, func() {
	fmt.Println("RunAfter")
})

time.Sleep(2500 * time.Millisecond)

Output:

RunAfter

func (*Server) RunEvery

func (s *Server) RunEvery(d time.Duration, f func()) *timingwheel.Timer

RunEvery：定时任务，定时每 Duration 时间执行 f

Example

handler := new(example)

s, err := NewServer(handler,
	Network("tcp"),
	Address(":1833"),
	NumLoops(8),
	ReusePort(true))
if err != nil {
	panic(err)
}

go s.Start()
defer s.Stop()

t := s.RunEvery(time.Second, func() {
	fmt.Println("EveryFunc")
})

time.Sleep(4500 * time.Millisecond)
t.Stop()
time.Sleep(4500 * time.Millisecond)

Output:

EveryFunc
EveryFunc
EveryFunc
EveryFunc

func (*Server) Start

func (s *Server) Start()

Start：启动 Server

func (*Server) Stop

func (s *Server) Stop()

Stop：关闭 Server