bingyi/gf
1
0
Fork 0
mirror of https://gitee.com/johng/gf synced 2026-06-29 02:26:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

ghttp.Server热重启特性开发中

Browse Source
This commit is contained in:
John
2018-05-10 23:52:09 +08:00
parent 51afecd675
commit 7ed17b5eb1
5 changed files with 154 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
g/net/ghttp/ghttp_server.go
View File

@ -19,14 +19,9 @@ import (
"gitee.com/johng/gf/g/container/gmap"
"gitee.com/johng/gf/g/container/gtype"
"gitee.com/johng/gf/g/container/gqueue"
"fmt"
"net"
"syscall"
"gitee.com/johng/gf/g/os/gcmd"
"gitee.com/johng/gf/g/os/gproc"
"gitee.com/johng/gf/g/encoding/gjson"
"gitee.com/johng/gf/g/os/gtime"
"time"
)
const (
@ -38,7 +33,6 @@ const (
gDEFAULT_COOKIE_MAX_AGE = 86400*365 // 默认cookie有效期(一年)
gDEFAULT_SESSION_MAX_AGE = 600 // 默认session有效期(600秒)
gDEFAULT_SESSION_ID_NAME = "gfsessionid" // 默认存放Cookie中的SessionId名称
gDEFAULT_COMMAND_PORT = 336816 // 默认本地命令控制端口
)
// ghttp.Server结构体
@ -48,7 +42,6 @@ type Server struct {
config ServerConfig // 配置对象
status int8 // 当前服务器状态(0：未启动，1：运行中)
servers []*gracefulServer // 底层http.Server列表
cmdPort int // 本地Web Server命令控制端口
methodsMap map[string]bool // 所有支持的HTTP Method(初始化时自动填充)
servedCount *gtype.Int // 已经服务的请求数(4-8字节，不考虑溢出情况)，同时作为请求ID
closeQueue *gqueue.Queue // 请求结束的关闭队列(存放的是需要异步关闭处理的*Request对象)
@ -81,8 +74,6 @@ type Server struct {
accessLogEnabled *gtype.Bool // 是否开启access log
accessLogger *glog.Logger // access log日志对象
errorLogger *glog.Logger // error log日志对象
// 多进程管理控制
manager *gproc.Manager // 多进程管理
}
// 域名、URI与回调函数的绑定记录表
@ -107,9 +98,61 @@ type HandlerItem struct {
// http注册函数
type HandlerFunc func(r *Request)
// 文件描述符map
type listenerFdMap map[string]string
// Server表，用以存储和检索名称与Server对象之间的关联关系
var serverMapping = gmap.NewStringInterfaceMap()
// Web Server多进程管理器
var procManager = gproc.NewManager()
// Web Server开始执行事件通道，由于同一个进程支持多Server，因此该通道为非阻塞
var runChan = make(chan struct{}, 100000)
// 已完成的run消息队列
var doneChan = make(chan struct{}, 100000)
// Web Server进程初始化
func init() {
go func() {
<- runChan
// 主进程只负责创建子进程
if !gproc.IsChild() {
sendProcessMsg(os.Getpid(), gMSG_START, nil)
}
// 开启进程消息监听处理
handleProcessMsg()
doneChan <- struct{}{}
}()
}
// 获取所有Web Server的文件描述符map
func getServerFdMap() map[string]listenerFdMap {
sfm := make(map[string]listenerFdMap)
serverMapping.RLockFunc(func(m map[string]interface{}) {
for k, v := range m {
sfm[k] = v.(*Server).getListenerFdMap()
}
})
return sfm
}
// 二进制转换为FdMap
func bufferToServerFdMap(buffer []byte) map[string]listenerFdMap {
sfm := make(map[string]listenerFdMap)
if len(buffer) > 0 {
j, _ := gjson.LoadContent(buffer, "json")
for k, _ := range j.ToMap() {
m := make(map[string]string)
for k, v := range j.GetMap(k) {
m[k] = gconv.String(v)
}
sfm[k] = m
}
}
return sfm
}
// 获取/创建一个默认配置的HTTP Server(默认监听端口是80)
// 单例模式，请保证name的唯一性
func GetServer(name...interface{}) (*Server) {
@ -123,7 +166,6 @@ func GetServer(name...interface{}) (*Server) {
s := &Server {
name : sname,
servers : make([]*gracefulServer, 0),
cmdPort : gDEFAULT_COMMAND_PORT,
methodsMap : make(map[string]bool),
handlerMap : make(HandlerMap),
statusHandlerMap : make(map[string]HandlerFunc),
@ -145,7 +187,6 @@ func GetServer(name...interface{}) (*Server) {
accessLogger : glog.New(),
errorLogger : glog.New(),
logHandler : gtype.NewInterface(),
manager : gproc.NewManager(),
}
s.errorLogger.SetBacktraceSkip(4)
s.accessLogger.SetBacktraceSkip(4)
@ -162,6 +203,13 @@ func GetServer(name...interface{}) (*Server) {
// 阻塞执行监听
func (s *Server) Run() error {
runChan <- struct{}{}
if !gproc.IsChild() {
<- doneChan
return nil
}
if s.status == 1 {
return errors.New("server is already running")
}
@ -174,24 +222,12 @@ func (s *Server) Run() error {
// 开启异步关闭队列处理循环
s.startCloseQueueLoop()
// 主进程只负责创建子进程
if !gproc.IsChild() {
p := s.manager.NewProcess(os.Args[0], os.Args, os.Environ())
p.Run()
gtime.SetTimeout(3*time.Second, func() {
b, _ := gjson.Encode(s.getAllListenerFdMap())
s.sendMsg(p.Pid(), gMSG_START, b)
})
}
// 开启进程消息监听处理
s.handleProcessMsg()
<- doneChan
return nil
}
// 开启底层Web Server执行
func (s *Server) startServer(fdMap map[string]string) {
fmt.Println("startServer")
func (s *Server) startServer(fdMap listenerFdMap) {
// 开始执行底层Web Server创建，端口监听
var wg sync.WaitGroup
var server *gracefulServer
@ -223,6 +259,7 @@ func (s *Server) startServer(fdMap map[string]string) {
server = s.newGracefulServer(item)
}
s.servers = append(s.servers, server)
glog.Printfln("https server started listening on %s", server.addr)
if err := server.ListenAndServeTLS(s.config.HTTPSCertPath, s.config.HTTPSKeyPath); err != nil {
// 如果非关闭错误，那么提示报错，否则认为是正常的服务关闭操作
if !strings.EqualFold(http.ErrServerClosed.Error(), err.Error()) {
@ -255,6 +292,7 @@ func (s *Server) startServer(fdMap map[string]string) {
server = s.newGracefulServer(item)
}
s.servers = append(s.servers, server)
glog.Printfln("http server started listening on %s", server.addr)
if err := server.ListenAndServe(); err != nil {
// 如果非关闭错误，那么提示报错，否则认为是正常的服务关闭操作
if !strings.EqualFold(http.ErrServerClosed.Error(), err.Error()) {
@ -274,23 +312,23 @@ func (s *Server) startServer(fdMap map[string]string) {
// 重启Web Server
func (s *Server) Restart() {
// 如果是主进程，那么向所有子进程发送重启信号
if !s.isChildProcess() {
s.manager.SignalAll(syscall.SIGUSR2)
if !gproc.IsChild() {
return
}
if pid, err := s.forkChildProcess(); err != nil {
glog.Errorf("server restart failed: %v, continue serving\n", err)
} else {
glog.Printf("server restart successfully, new pid: %d\n", pid)
s.Shutdown()
}
//if pid, err := s.forkChildProcess(); err != nil {
// glog.Errorf("server restart failed: %v, continue serving\n", err)
//} else {
// glog.Printf("server restart successfully, new pid: %d\n", pid)
// s.Shutdown()
//}
}
// 关闭Web Server
func (s *Server) Shutdown() {
// 如果是主进程，那么向所有子进程发送关闭信号
if !s.isChildProcess() {
s.manager.SignalAll(syscall.SIGTERM)
if !gproc.IsChild() {
return
}
for _, v := range s.servers {
@ -298,27 +336,10 @@ func (s *Server) Shutdown() {
}
}
// 子进程获取的文件打开数
func (s *Server) processFileCount() int {
return gconv.Int(gcmd.Option.Get("fcount"))
}
// 判断是否为子进程执行
func (s *Server) isChildProcess() bool {
return s.getTopId() > 0
}
// 获取顶级进程ID(管理进程ID)
func (s *Server) getTopId() int {
id := gcmd.Option.Get("topid")
if id != "" {
return gconv.Int(id)
}
return 0
}
// 获取当前监听的文件描述符信息，构造成map返回
func (s *Server) getAllListenerFdMap() map[string]string {
func (s *Server) getListenerFdMap() map[string]string {
m := map[string]string{
"http" : "",
"https" : "",
@ -344,46 +365,6 @@ func (s *Server) getAllListenerFdMap() map[string]string {
return m
}
// 二进制转换为FdMap
func (s *Server) bufferToFdMap(buffer []byte) map[string]string {
m := make(map[string]string)
j, _ := gjson.LoadContent(buffer, "json")
for k, v := range j.ToMap() {
m[k] = gconv.String(v)
}
return m
}
// 创建子进程来监听并处理新的HTTP请求，与父进程使用的是同一个socket文件描述符
func (s *Server) forkChildProcess() (int, error) {
// 获取所有http server的file
files := []*os.File{os.Stdin,os.Stdout,os.Stderr}
for _, v := range s.servers {
if f, e := v.listener.(*net.TCPListener).File(); e == nil {
files = append(files, f)
} else {
return 0, fmt.Errorf("failed to get listener file: %v", e)
}
}
// 开启子进程，并传递socket文件指针
topId := s.getTopId()
if topId == 0 {
topId = os.Getpid()
}
args := make([]string, 4)
args[0] = os.Args[0]
args[1] = fmt.Sprintf("--name=%s", s.name)
args[2] = fmt.Sprintf("--port=%d", s.cmdPort)
args[3] = fmt.Sprintf("--fcount=%d", len(files) - 3)
p := s.manager.NewProcess(os.Args[0], args, os.Environ())
p.GetAttr().Files = files
if pid, err := p.Run(); err != nil {
return 0, fmt.Errorf("failed to fork process: %v", err)
} else {
return pid, nil
}
}
// 清空当前的handlerCache
func (s *Server) clearHandlerCache() {
s.hmcmu.Lock()

52
g/net/ghttp/ghttp_server_comm.go
View File

@ -8,23 +8,23 @@
package ghttp
import (
"os"
"gitee.com/johng/gf/g/os/gproc"
"gitee.com/johng/gf/g/encoding/gbinary"
"fmt"
"gitee.com/johng/gf/g/encoding/gjson"
"os"
)
const (
gMSG_START = iota
gMSG_RESTART
gMSG_SHUTDOWN
gMSG_EXIT
)
// 处理进程间消息
// 数据格式： 操作(8bit) | 参数(变长)
func (s *Server) handleProcessMsg() {
func handleProcessMsg() {
for {
if msg := gproc.Receive(); msg != nil {
fmt.Println(msg)
@ -32,19 +32,48 @@ func (s *Server) handleProcessMsg() {
data := msg.Data[1 : ]
if gproc.IsChild() {
switch act {
// 开启所有Web Server(根据消息启动)
case gMSG_START:
s.startServer(s.bufferToFdMap(data))
sfm := bufferToServerFdMap(data)
for k, v := range sfm {
GetServer(k).startServer(v)
}
// 子进程收到重启消息，那么将自身的ServerFdMap信息收集后发送给主进程，由主进程进行统一调度
case gMSG_RESTART:
case gMSG_SHUTDOWN: s.Shutdown()
case gMSG_EXIT: os.Exit(0)
b, _ := gjson.Encode(getServerFdMap())
sendProcessMsg(gproc.Ppid(), gMSG_RESTART, b)
// 友好关闭服务链接并退出
case gMSG_SHUTDOWN:
serverMapping.RLockFunc(func(m map[string]interface{}) {
for _, v := range m {
v.(*Server).Shutdown()
}
})
return
}
} else {
switch act {
// 开启服务
case gMSG_START:
p := procManager.NewProcess(os.Args[0], os.Args, os.Environ())
p.Run()
sendProcessMsg(p.Pid(), gMSG_START, nil)
// 重启服务
case gMSG_RESTART:
// 创建新的服务进程，使用文件描述来监听同样的端口
p := procManager.NewProcess(os.Args[0], os.Args, os.Environ())
p.Run()
sendProcessMsg(p.Pid(), gMSG_START, data)
// 关闭旧的服务进程
sendProcessMsg(msg.Pid, gMSG_SHUTDOWN, nil)
// 关闭服务
case gMSG_SHUTDOWN:
case gMSG_EXIT: os.Exit(0)
procManager.Send(formatMsgBuffer(gMSG_SHUTDOWN, nil))
}
}
@ -53,7 +82,12 @@ func (s *Server) handleProcessMsg() {
}
// 向进程发送操作消息
func (s *Server) sendMsg(pid int, act int, data []byte) {
gproc.Send(pid, append(gbinary.EncodeInt8(int8(act)), data...))
func sendProcessMsg(pid int, act int, data []byte) {
gproc.Send(pid, formatMsgBuffer(act, data))
}
// 生成一条满足Web Server进程通信协议的消息
func formatMsgBuffer(act int, data []byte) []byte {
return append(gbinary.EncodeInt8(int8(act)), data...)
}

4
g/os/gfile/gfile.go
View File

@ -46,6 +46,10 @@ func Mkdir(path string) error {
// 给定文件的绝对路径创建文件
func Create(path string) error {
dir := Dir(path)
if !Exists(dir) {
Mkdir(dir)
}
f, err := os.Create(path)
if err != nil {
return err

35
g/os/gproc/gproc_comm.go
View File

@ -16,6 +16,7 @@ import (
"gitee.com/johng/gf/g/os/gfsnotify"
"gitee.com/johng/gf/g/container/gqueue"
"gitee.com/johng/gf/g/encoding/gbinary"
"gitee.com/johng/gf/g/os/gtime"
)
const (
@ -38,25 +39,39 @@ type Msg struct {
func init() {
path := getCommFilePath(os.Getpid())
if !gfile.Exists(path) {
gfile.Create(path)
if err := gfile.Create(path); err != nil {
glog.Error(err)
}
} else {
// 初始化时读取已有数据(文件修改时间在10秒以内)
if gtime.Second() - gfile.MTime(path) < 10 {
checkCommBuffer(path)
}
}
fmt.Println(path)
// 文件事件监听，如果通信数据文件有任何变化，读取文件并添加到消息队列
err := gfsnotify.Add(path, func(event *gfsnotify.Event) {
fmt.Println(event)
commLocker.Lock()
buffer := gfile.GetBinContents(path)
os.Truncate(path, 0)
commLocker.UnLock()
for _, v := range bufferToMsgs(buffer) {
commQueue.PushBack(v)
}
checkCommBuffer(path)
})
if err != nil {
glog.Error(err)
}
}
// 手动检查进程通信消息，如果存在消息曾推送到进程消息队列
func checkCommBuffer(path string) {
commLocker.Lock()
buffer := gfile.GetBinContents(path)
if len(buffer) > 0 {
os.Truncate(path, 0)
}
commLocker.UnLock()
if len(buffer) > 0 {
for _, v := range bufferToMsgs(buffer) {
commQueue.PushBack(v)
}
}
}
// 获取其他进程传递到当前进程的消息包，阻塞执行
func Receive() *Msg {
if v := commQueue.PopFront(); v != nil {

64
geg/other/test.go
View File

@ -2,62 +2,14 @@ package main
import (
"fmt"
"net/http"
"github.com/tabalt/gracehttp"
"gitee.com/johng/gf/g/encoding/gbinary"
"gitee.com/johng/gf/g/os/gproc"
"gitee.com/johng/gf/g/os/gfile"
"gitee.com/johng/gf/g/os/gtime"
)
func test() {
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
fmt.Fprintf(w, "hello world")
})
s := gracehttp.NewServer(":8888", nil, gracehttp.DEFAULT_READ_TIMEOUT, gracehttp.DEFAULT_WRITE_TIMEOUT)
err := s.ListenAndServe()
if err != nil {
fmt.Println(err)
}
}
// 常见的二进制数据校验方式，生成校验结果
func checksum(buffer []byte) uint32 {
var checksum uint32
for _, b := range buffer {
checksum += uint32(b)
}
return checksum
}
// 数据解包，防止黏包
func bufferToMsgs(buffer []byte) []*gproc.Msg {
s := 0
msgs := make([]*gproc.Msg, 0)
for s < len(buffer) {
fmt.Println(s)
length := gbinary.DecodeToInt(buffer[s : s + 4])
if length < 0 || length > len(buffer) {
s++
continue
}
checksum1 := gbinary.DecodeToUint32(buffer[s + 8 : s + 12])
checksum2 := checksum(buffer[s + 12 : s + length])
if checksum1 != checksum2 {
s++
continue
}
msgs = append(msgs, &gproc.Msg {
Pid : gbinary.DecodeToInt(buffer[s + 4 : s + 8]),
Data : buffer[s + 12 : s + length],
})
s += length
}
return msgs
}
func main() {
b := []byte{26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33, 26, 0, 0, 0, 60, 109, 0, 0, 84, 5, 0, 0, 104, 101, 108, 108, 111, 32, 112, 114, 111, 99, 101, 115, 115, 33}
m := bufferToMsgs(b)
fmt.Println(m)
}
func main(){
t1 := gfile.MTime("/home/john/Workspace/Go/GOPATH/src/gitee.com/johng/gf/geg/other/test.go")
t2 := gtime.Second()
fmt.Println(t1)
fmt.Println(t2)
}