proxy

README

proxy

tcpproxy

#可以通过下面的命令下载

go get github.com/mx5566/proxy

1. tcp代理
2. 采用一致性哈希算法进行负载均衡
3. 后端服务器心跳检查有tcp检查或者http检查、支持失败次数、成功次数、连接超时、允许设置初始的服务器状态、修改服务器心跳检测的端口 如果设置为0就采用后端服务器端口
4. 可以通过stats 获取后端服务器的状态列表
5. 日志采用的是uber的zap
6. 增加限流器 总共有三个

限流器

const (
	queueLimter        LimitType = iota + 1 // 队列模式限流
	tokenBucketLimter                       // golang官方库  golang.org/x/time/rate  bucket
	slideWindowLimiter                      // 滑动窗口限流器 tcp滑动窗口 -- 没有实现
	leakBucketLimter                        // 漏斗桶限流器
)



1对于队列模式queueLimter
采用的是二级模式
配置文件中
  wait_queue_len: 100 # 代表可以处于等待处理的队列
  max_conn: 50000 # 代表同时处于处理的连接 并发的用户

函数分为三步
-- NewQueueLimter(wait_queue_len, max_conn)
-- Bind(handler func(conn interface{}))
-- 		if this.limiter.IsAvalivale() {
   			this.limiter.SetWaitQueue(conn)
   		} 



2 tokenBucketLimter
  duration: 8  # 单位毫秒--速率
  captity: 100 # 容量
函数分为三步
-- NewTokenBucketLimiter(duration, captity)
-- Bind(handler func(conn interface{}))
-- 		if this.limiter.IsAvalivale() {
   			this.limiter.SetWaitQueue(conn)
   		} 

3 LeakBucketLimiter
  duration: 8  # 单位毫秒--速率
  captity: 100 # 容量
  name: "Test" # 限流器名字

函数分为三步
-- NewLeakBucketLimiter(name, captity, duration)
-- Bind(handler func(conn interface{}))
-- 		if this.limiter.IsAvalivale() {
   			this.limiter.SetWaitQueue(conn)
   		} 

例子直接通过调用下面的函数就可以 里面的yaml文件目录可以自己设定

    CreateProxy("./config.yaml")

Documentation

Index

• func CreateProxy(configPath string)
• func NewLogger(filePath string, level zapcore.Level, maxSize int, maxBackups int, maxAge int, ...) *zap.Logger
• func StatsHandler(w http.ResponseWriter, r *http.Request)
• type BackStats
• type BackendEnd
• type Banlance
  • func (this *Banlance) GetBackEndServer(conn net.Conn) string
  • func (this *Banlance) Init(backend []string, heatch HeatchConfig)
  • func (this *Banlance) ServiceGovernance(stats []BackStats)
• type Context
  • func (this *Context) Header(key, value string)
  • func (this *Context) Reset(rw http.ResponseWriter, r *http.Request)
  • func (this *Context) ServerJson(data interface{})
  • func (this *Context) WriteString(content string)
• type HeatchConfig
• type HeathMontior
  • func (this *HeathMontior) HttpCheck(hConfig HeatchConfig, backend map[string]*BackendEnd)
  • func (this *HeathMontior) ParseIP(port int, svrStr string) string
  • func (this *HeathMontior) TcpCheck(hConfig HeatchConfig, backend map[string]*BackendEnd)
• type LeakBucketLimiter
  • func NewLeakBucketLimiter(name string, cap uint, t int) *LeakBucketLimiter
  • func (this *LeakBucketLimiter) Bind(handler func(conn interface{}))
  • func (this *LeakBucketLimiter) IsAvalivale() bool
  • func (this *LeakBucketLimiter) SetWaitQueue(conn interface{})
• type LimitInterface
  • func CreateLimiter(lConfig LimiterConfig) LimitInterface
• type LimitType
• type LimiterConfig
• type LogConfig
• type Proxy
  • func (this *Proxy) Copy(from net.Conn, to net.Conn, ok chan bool)
  • func (this *Proxy) HandleConnect(conn interface{})
  • func (this *Proxy) InitProxy(proxyConfig *ProxyConfig)
  • func (this *Proxy) OnSignalExit()
  • func (this *Proxy) StartBanlance(proxyConfig *ProxyConfig)
  • func (this *Proxy) StartStat()
• type ProxyConfig
• type QueueLimiter
  • func NewQueueLimter(waitLength, maxConn int) *QueueLimiter
  • func (this QueueLimiter) Bind(handler func(conn interface{}))
  • func (this QueueLimiter) IsAvalivale() bool
  • func (this *QueueLimiter) SetWaitQueue(conn interface{})
• type Stat
  • func (this *Stat) RegisterRoute(uri string, f func(w http.ResponseWriter, r *http.Request))
  • func (this *Stat) StartStat()
• type TokenBucketLimiter
  • func NewTokenBucketLimiter(t int, token int) *TokenBucketLimiter
  • func (this *TokenBucketLimiter) Bind(handler func(conn interface{}))
  • func (this *TokenBucketLimiter) IsAvalivale() bool
  • func (this *TokenBucketLimiter) SetWaitQueue(conn interface{})

Functions

func CreateProxy

func CreateProxy(configPath string)

func NewLogger

func NewLogger(filePath string, level zapcore.Level, maxSize int, maxBackups int, maxAge int, compress bool, serviceName string) *zap.Logger

func StatsHandler

func StatsHandler(w http.ResponseWriter, r *http.Request)

监控状态处理函数 所有后端服务器的状态

Types

type BackStats

type BackStats struct {
	IsUp   bool // is Up or Down
	SvrStr string
}

type BackendEnd

type BackendEnd struct {
	SvrStr    string `json:"svrStr"`
	IsUp      bool   `json:"isUp"`      // is Up or Down
	FailTimes int    `json:"failTimes"` // 失败次数
	RiseTimes int    `json:"riseTimes"` // 连接成功的次数
}

BackendSvr Type

type Banlance

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

func (*Banlance) GetBackEndServer

func (this *Banlance) GetBackEndServer(conn net.Conn) string

获得后端服务器

func (*Banlance) Init

func (this *Banlance) Init(backend []string, heatch HeatchConfig)

func (*Banlance) ServiceGovernance

func (this *Banlance) ServiceGovernance(stats []BackStats)

后端应用服务器服务治理

type Context

type Context struct {
	Request        *http.Request
	ResponseWriter http.ResponseWriter
}

func (*Context) Header

func (this *Context) Header(key, value string)

func (*Context) Reset

func (this *Context) Reset(rw http.ResponseWriter, r *http.Request)

重置

func (*Context) ServerJson

func (this *Context) ServerJson(data interface{})

func (*Context) WriteString

func (this *Context) WriteString(content string)

响应字符串

type HeatchConfig

type HeatchConfig struct {
	Interval             int      `yaml:"interval"`
	Rise                 int      `yaml:"rise"`
	Fall                 int      `yaml:"fall"`
	Timeout              int      `yaml:"timeout"`
	Type                 string   `yaml:"type"`
	DefaultDown          bool     `yaml:"default_down"`
	CheckHttpSend        string   `yaml:"check_http_send"`
	CheckHttpExceptAlive []string `yaml:"check_http_expect_alive"`
	Port                 int      `yaml:"port"`
}

HeatchConfig Type

type HeathMontior

type HeathMontior struct {
}

HeatchMontior

func (*HeathMontior) HttpCheck

func (this *HeathMontior) HttpCheck(hConfig HeatchConfig, backend map[string]*BackendEnd)

HttpCheck

func (*HeathMontior) ParseIP

func (this *HeathMontior) ParseIP(port int, svrStr string) string

func (*HeathMontior) TcpCheck

func (this *HeathMontior) TcpCheck(hConfig HeatchConfig, backend map[string]*BackendEnd)

TcpCheck

type LeakBucketLimiter

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

////////////////////////////LeakBucketLimiter///////////////////////////// 漏斗桶限流算法

func NewLeakBucketLimiter

func NewLeakBucketLimiter(name string, cap uint, t int) *LeakBucketLimiter

func (*LeakBucketLimiter) Bind

func (this *LeakBucketLimiter) Bind(handler func(conn interface{}))

bind handler function to handler

func (*LeakBucketLimiter) IsAvalivale

func (this *LeakBucketLimiter) IsAvalivale() bool

func (*LeakBucketLimiter) SetWaitQueue

func (this *LeakBucketLimiter) SetWaitQueue(conn interface{})

call handler function by conn

type LimitInterface

type LimitInterface interface {
	IsAvalivale() bool
	Bind(handler func(conn interface{}))
	SetWaitQueue(conn interface{})
}

限流接口

func CreateLimiter

func CreateLimiter(lConfig LimiterConfig) LimitInterface

不同的限流器的初始化接口

type LimitType

type LimitType int

type LimiterConfig

type LimiterConfig struct {
	Type         int    `yaml:"type"`
	WaitQueueLen int    `yaml:"wait_queue_len"`
	MaxConn      int    `yaml:"max_conn"`
	Duration     int    `yaml:"duration"`
	Captity      uint   `yaml:"captity"`
	Name         string `yaml:"name"`
}

type LogConfig

type LogConfig struct {
	Level       int8   `yaml:"level"`
	Path        string `yaml:"path"`
	MaxSize     int    `yaml:"max_size"`
	MaxBackup   int    `yaml:"max_backup"`
	MaxAge      int    `yaml:"max_age"`
	Compress    bool   `yaml:"compress"`
	ServiceName string `yaml:"servicename"`
}

LogConfig Type

type Proxy

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

func (*Proxy) Copy

func (this *Proxy) Copy(from net.Conn, to net.Conn, ok chan bool)

func (*Proxy) HandleConnect

func (this *Proxy) HandleConnect(conn interface{})

func (*Proxy) InitProxy

func (this *Proxy) InitProxy(proxyConfig *ProxyConfig)

func (*Proxy) OnSignalExit

func (this *Proxy) OnSignalExit()

信号处理

func (*Proxy) StartBanlance

func (this *Proxy) StartBanlance(proxyConfig *ProxyConfig)

func (*Proxy) StartStat

func (this *Proxy) StartStat()

type ProxyConfig

type ProxyConfig struct {
	Bind    string        `yaml:"bind"`
	Backend []string      `yaml:"backend"`
	Log     LogConfig     `yaml:"log"`
	Stats   string        `yaml:"stats"`
	Heatch  HeatchConfig  `yaml:"heatch"`
	Limter  LimiterConfig `yaml:"limiter"`
}

ProxyConfig Type

type QueueLimiter

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

/////////////////////////////QueueLimiter///////////////////////////// 通过队列实现限流

func NewQueueLimter

func NewQueueLimter(waitLength, maxConn int) *QueueLimiter

NewQueueLimter *

waitLength-最大的等待处理的长度
maxConn-最大的并发处理长度

*

func (QueueLimiter) Bind

func (this QueueLimiter) Bind(handler func(conn interface{}))

func (QueueLimiter) IsAvalivale

func (this QueueLimiter) IsAvalivale() bool

等待队列是否还有空位

func (*QueueLimiter) SetWaitQueue

func (this *QueueLimiter) SetWaitQueue(conn interface{})

限流器增加计数

type Stat

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

Stat Real Server States

func (*Stat) RegisterRoute

func (this *Stat) RegisterRoute(uri string, f func(w http.ResponseWriter, r *http.Request))

注册路由处理函数

func (*Stat) StartStat

func (this *Stat) StartStat()

type TokenBucketLimiter

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

令牌桶实现限流

func NewTokenBucketLimiter

func NewTokenBucketLimiter(t int, token int) *TokenBucketLimiter

func (*TokenBucketLimiter) Bind

func (this *TokenBucketLimiter) Bind(handler func(conn interface{}))

bind handler function to handler

func (*TokenBucketLimiter) IsAvalivale

func (this *TokenBucketLimiter) IsAvalivale() bool

func (*TokenBucketLimiter) SetWaitQueue

func (this *TokenBucketLimiter) SetWaitQueue(conn interface{})

call handler function by conn