添加定时器, 检查心跳

zhangqijia
1 parent 98b0736d
Showing 19 changed files with 377 additions and 779 deletions Show diff stats
cmd/game.go
protos
src/common/common.go
src/components/kafkatimer/delayqueue.go
src/components/kafkatimer/delayqueue/delayqueue.go
src/components/kafkatimer/timingwheel.go
src/components/kafkatimer/timingwheel_test.go
src/components/kafkatimer/utils.go
src/components/net/conn.go
src/components/net/server.go
src/components/skynettimer/timerwheel.go
src/components/kafkatimer/bucket.go -> src/components/timewheel/bucket.go
src/components/timewheel/timewheel.go
src/components/timewheel/timewheel_test.go
src/components/workpool/workpool.go
src/plugin/RolePlugin.go
src/plugin/protocode.go
src/utils/utils.go
test/client.go
@@ -6,6 +6,7 @@ import (
 	"pro2d/conf"
 	"pro2d/src/components/logger"
 	"pro2d/src/components/net"
+	_ "pro2d/src/plugin"
 	"syscall"
 )
-Subproject commit c29c12e0a07ed8e3a35a451ca7dbdadbc4bb70fb
+Subproject commit 3aa479358834e2ea24c1c1b75c739b0524de1c4f
 package common
 const (
+	//协程池 大小
 	WorkerPoolSize = 10
 	MaxTaskPerWorker = 100
-	MaxPacketSize = 10 * 1024 * 1024
-	MaxMsgChanLen = 20
-)
-const (
+	//包头
 	HEADLEN = 16
+	//jwt
 	Pro2DTokenSignedString = "Pro2DSecret"
+
+	//定时器
+	TickMS = 10
+	WheelSize = 3600
+
+	//心跳
+	HEART_TIMER_INTERVAL = 5 //s
+	HEART_TIMEOUT_COUNT_MAX = 20 //最大超时次数
 )
@@ -1,183 +0,0 @@
-package kafkatimer
-
-import (
-	"container/heap"
-	"sync"
-	"sync/atomic"
-	"time"
-)
-
-type item struct {
-	Value    interface{}
-	Priority int64
-	Index    int
-}
-
-// this is a priority queue as implemented by a min heap
-// ie. the 0th element is the *lowest* value
-type priorityQueue []*item
-
-func newPriorityQueue(capacity int) priorityQueue {
-	return make(priorityQueue, 0, capacity)
-}
-
-func (pq priorityQueue) Len() int {
-	return len(pq)
-}
-
-func (pq priorityQueue) Less(i, j int) bool {
-	return pq[i].Priority < pq[j].Priority
-}
-
-func (pq priorityQueue) Swap(i, j int) {
-	pq[i], pq[j] = pq[j], pq[i]
-	pq[i].Index = i
-	pq[j].Index = j
-}
-
-func (pq *priorityQueue) Push(x interface{}) {
-	n := len(*pq)
-	c := cap(*pq)
-	if n+1 > c {
-		npq := make(priorityQueue, n, c*2)
-		copy(npq, *pq)
-		*pq = npq
-	}
-	*pq = (*pq)[0 : n+1]
-	item := x.(*item)
-	item.Index = n
-	(*pq)[n] = item
-}
-
-func (pq *priorityQueue) Pop() interface{} {
-	n := len(*pq)
-	c := cap(*pq)
-	if n < (c/2) && c > 25 {
-		npq := make(priorityQueue, n, c/2)
-		copy(npq, *pq)
-		*pq = npq
-	}
-	item := (*pq)[n-1]
-	item.Index = -1
-	*pq = (*pq)[0 : n-1]
-	return item
-}
-
-func (pq *priorityQueue) PeekAndShift(max int64) (*item, int64) {
-	if pq.Len() == 0 {
-		return nil, 0
-	}
-
-	item := (*pq)[0]
-	if item.Priority > max {
-		return nil, item.Priority - max
-	}
-	heap.Remove(pq, 0)
-
-	return item, 0
-}
-
-// The end of PriorityQueue implementation.
-
-// DelayQueue is an unbounded blocking queue of *Delayed* elements, in which
-// an element can only be taken when its delay has expired. The head of the
-// queue is the *Delayed* element whose delay expired furthest in the past.
-type DelayQueue struct {
-	C chan interface{}
-
-	mu sync.Mutex
-	pq priorityQueue
-
-	// Similar to the sleeping state of runtime.timers.
-	sleeping int32
-	wakeupC  chan struct{}
-}
-
-// New creates an instance of delayQueue with the specified size.
-func New(size int) *DelayQueue {
-	return &DelayQueue{
-		C:       make(chan interface{}),
-		pq:      newPriorityQueue(size),
-		wakeupC: make(chan struct{}),
-	}
-}
-
-// Offer inserts the element into the current queue.
-func (dq *DelayQueue) Offer(elem interface{}, expiration int64) {
-	item := &item{Value: elem, Priority: expiration}
-
-	dq.mu.Lock()
-	heap.Push(&dq.pq, item)
-	index := item.Index
-	dq.mu.Unlock()
-
-	if index == 0 {
-		// A new item with the earliest expiration is added.
-		if atomic.CompareAndSwapInt32(&dq.sleeping, 1, 0) {
-			dq.wakeupC <- struct{}{}
-		}
-	}
-}
-
-// Poll starts an infinite loop, in which it continually waits for an element
-// to expire and then send the expired element to the channel C.
-func (dq *DelayQueue) Poll(exitC chan struct{}, nowF func() int64) {
-	for {
-		now := nowF()
-
-		dq.mu.Lock()
-		item, delta := dq.pq.PeekAndShift(now)
-		if item == nil {
-			// No items left or at least one item is pending.
-
-			// We must ensure the atomicity of the whole operation, which is
-			// composed of the above PeekAndShift and the following StoreInt32,
-			// to avoid possible race conditions between Offer and Poll.
-			atomic.StoreInt32(&dq.sleeping, 1)
-		}
-		dq.mu.Unlock()
-
-		if item == nil {
-			if delta == 0 {
-				// No items left.
-				select {
-				case <-dq.wakeupC:
-					// Wait until a new item is added.
-					continue
-				case <-exitC:
-					goto exit
-				}
-			} else if delta > 0 {
-				// At least one item is pending.
-				select {
-				case <-dq.wakeupC:
-					// A new item with an "earlier" expiration than the current "earliest" one is added.
-					continue
-				case <-time.After(time.Duration(delta) * time.Millisecond):
-					// The current "earliest" item expires.
-
-					// Reset the sleeping state since there's no need to receive from wakeupC.
-					if atomic.SwapInt32(&dq.sleeping, 0) == 0 {
-						// A caller of Offer() is being blocked on sending to wakeupC,
-						// drain wakeupC to unblock the caller.
-						<-dq.wakeupC
-					}
-					continue
-				case <-exitC:
-					goto exit
-				}
-			}
-		}
-
-		select {
-		case dq.C <- item.Value:
-			// The expired element has been sent out successfully.
-		case <-exitC:
-			goto exit
-		}
-	}
-
-exit:
-	// Reset the states
-	atomic.StoreInt32(&dq.sleeping, 0)
-}
@@ -1,186 +0,0 @@
-package delayqueue
-
-import (
-	"container/heap"
-	"sync"
-	"sync/atomic"
-	"time"
-)
-
-// The start of PriorityQueue implementation.
-// Borrowed from https://github.com/nsqio/nsq/blob/master/internal/pqueue/pqueue.go
-
-type item struct {
-	Value    interface{}
-	Priority int64
-	Index    int
-}
-
-// this is a priority queue as implemented by a min heap
-// ie. the 0th element is the *lowest* value
-type priorityQueue []*item
-
-func newPriorityQueue(capacity int) priorityQueue {
-	return make(priorityQueue, 0, capacity)
-}
-
-func (pq priorityQueue) Len() int {
-	return len(pq)
-}
-
-func (pq priorityQueue) Less(i, j int) bool {
-	return pq[i].Priority < pq[j].Priority
-}
-
-func (pq priorityQueue) Swap(i, j int) {
-	pq[i], pq[j] = pq[j], pq[i]
-	pq[i].Index = i
-	pq[j].Index = j
-}
-
-func (pq *priorityQueue) Push(x interface{}) {
-	n := len(*pq)
-	c := cap(*pq)
-	if n+1 > c {
-		npq := make(priorityQueue, n, c*2)
-		copy(npq, *pq)
-		*pq = npq
-	}
-	*pq = (*pq)[0 : n+1]
-	item := x.(*item)
-	item.Index = n
-	(*pq)[n] = item
-}
-
-func (pq *priorityQueue) Pop() interface{} {
-	n := len(*pq)
-	c := cap(*pq)
-	if n < (c/2) && c > 25 {
-		npq := make(priorityQueue, n, c/2)
-		copy(npq, *pq)
-		*pq = npq
-	}
-	item := (*pq)[n-1]
-	item.Index = -1
-	*pq = (*pq)[0 : n-1]
-	return item
-}
-
-func (pq *priorityQueue) PeekAndShift(max int64) (*item, int64) {
-	if pq.Len() == 0 {
-		return nil, 0
-	}
-
-	item := (*pq)[0]
-	if item.Priority > max {
-		return nil, item.Priority - max
-	}
-	heap.Remove(pq, 0)
-
-	return item, 0
-}
-
-// The end of PriorityQueue implementation.
-
-// DelayQueue is an unbounded blocking queue of *Delayed* elements, in which
-// an element can only be taken when its delay has expired. The head of the
-// queue is the *Delayed* element whose delay expired furthest in the past.
-type DelayQueue struct {
-	C chan interface{}
-
-	mu sync.Mutex
-	pq priorityQueue
-
-	// Similar to the sleeping state of runtime.timers.
-	sleeping int32
-	wakeupC  chan struct{}
-}
-
-// New creates an instance of delayQueue with the specified size.
-func New(size int) *DelayQueue {
-	return &DelayQueue{
-		C:       make(chan interface{}),
-		pq:      newPriorityQueue(size),
-		wakeupC: make(chan struct{}),
-	}
-}
-
-// Offer inserts the element into the current queue.
-func (dq *DelayQueue) Offer(elem interface{}, expiration int64) {
-	item := &item{Value: elem, Priority: expiration}
-
-	dq.mu.Lock()
-	heap.Push(&dq.pq, item)
-	index := item.Index
-	dq.mu.Unlock()
-
-	if index == 0 {
-		// A new item with the earliest expiration is added.
-		if atomic.CompareAndSwapInt32(&dq.sleeping, 1, 0) {
-			dq.wakeupC <- struct{}{}
-		}
-	}
-}
-
-// Poll starts an infinite loop, in which it continually waits for an element
-// to expire and then send the expired element to the channel C.
-func (dq *DelayQueue) Poll(exitC chan struct{}, nowF func() int64) {
-	for {
-		now := nowF()
-
-		dq.mu.Lock()
-		item, delta := dq.pq.PeekAndShift(now)
-		if item == nil {
-			// No items left or at least one item is pending.
-
-			// We must ensure the atomicity of the whole operation, which is
-			// composed of the above PeekAndShift and the following StoreInt32,
-			// to avoid possible race conditions between Offer and Poll.
-			atomic.StoreInt32(&dq.sleeping, 1)
-		}
-		dq.mu.Unlock()
-
-		if item == nil {
-			if delta == 0 {
-				// No items left.
-				select {
-				case <-dq.wakeupC:
-					// Wait until a new item is added.
-					continue
-				case <-exitC:
-					goto exit
-				}
-			} else if delta > 0 {
-				// At least one item is pending.
-				select {
-				case <-dq.wakeupC:
-					// A new item with an "earlier" expiration than the current "earliest" one is added.
-					continue
-				case <-time.After(time.Duration(delta) * time.Millisecond):
-					// The current "earliest" item expires.
-
-					// Reset the sleeping state since there's no need to receive from wakeupC.
-					if atomic.SwapInt32(&dq.sleeping, 0) == 0 {
-						// A caller of Offer() is being blocked on sending to wakeupC,
-						// drain wakeupC to unblock the caller.
-						<-dq.wakeupC
-					}
-					continue
-				case <-exitC:
-					goto exit
-				}
-			}
-		}
-
-		select {
-		case dq.C <- item.Value:
-			// The expired element has been sent out successfully.
-		case <-exitC:
-			goto exit
-		}
-	}
-
-exit:
-	// Reset the states
-	atomic.StoreInt32(&dq.sleeping, 0)
-}
@@ -1,227 +0,0 @@
-package kafkatimer
-
-import (
-	"errors"
-	"sync/atomic"
-	"time"
-	"unsafe"
-
-	"github.com/RussellLuo/timingwheel/delayqueue"
-)
-//时间轮 kafka
-
-// TimingWheel is an implementation of Hierarchical Timing Wheels.
-type TimingWheel struct {
-	tick      int64 // in milliseconds
-	wheelSize int64
-
-	interval    int64 // in milliseconds
-	currentTime int64 // in milliseconds
-	buckets     []*bucket
-	queue       *delayqueue.DelayQueue
-
-	// The higher-level overflow wheel.
-	//
-	// NOTE: This field may be updated and read concurrently, through Add().
-	overflowWheel unsafe.Pointer // type: *TimingWheel
-
-	exitC     chan struct{}
-	waitGroup waitGroupWrapper
-}
-
-// NewTimingWheel creates an instance of TimingWheel with the given tick and wheelSize.
-func NewTimingWheel(tick time.Duration, wheelSize int64) *TimingWheel {
-	tickMs := int64(tick / time.Millisecond)
-	if tickMs <= 0 {
-		panic(errors.New("tick must be greater than or equal to 1ms"))
-	}
-
-	startMs := timeToMs(time.Now().UTC())
-
-	return newTimingWheel(
-		tickMs,
-		wheelSize,
-		startMs,
-		delayqueue.New(int(wheelSize)),
-	)
-}
-
-// newTimingWheel is an internal helper function that really creates an instance of TimingWheel.
-func newTimingWheel(tickMs int64, wheelSize int64, startMs int64, queue *delayqueue.DelayQueue) *TimingWheel {
-	buckets := make([]*bucket, wheelSize)
-	for i := range buckets {
-		buckets[i] = newBucket()
-	}
-	return &TimingWheel{
-		tick:        tickMs,
-		wheelSize:   wheelSize,
-		currentTime: truncate(startMs, tickMs),
-		interval:    tickMs * wheelSize,
-		buckets:     buckets,
-		queue:       queue,
-		exitC:       make(chan struct{}),
-	}
-}
-
-// add inserts the timer t into the current timing wheel.
-func (tw *TimingWheel) add(t *Timer) bool {
-	currentTime := atomic.LoadInt64(&tw.currentTime)
-	if t.expiration < currentTime+tw.tick {
-		// Already expired
-		return false
-	} else if t.expiration < currentTime+tw.interval {
-		// Put it into its own bucket
-		virtualID := t.expiration / tw.tick
-		b := tw.buckets[virtualID%tw.wheelSize]
-		b.Add(t)
-
-		// Set the bucket expiration time
-		if b.SetExpiration(virtualID * tw.tick) {
-			// The bucket needs to be enqueued since it was an expired bucket.
-			// We only need to enqueue the bucket when its expiration time has changed,
-			// i.e. the wheel has advanced and this bucket get reused with a new expiration.
-			// Any further calls to set the expiration within the same wheel cycle will
-			// pass in the same value and hence return false, thus the bucket with the
-			// same expiration will not be enqueued multiple times.
-			tw.queue.Offer(b, b.Expiration())
-		}
-
-		return true
-	} else {
-		// Out of the interval. Put it into the overflow wheel
-		overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
-		if overflowWheel == nil {
-			atomic.CompareAndSwapPointer(
-				&tw.overflowWheel,
-				nil,
-				unsafe.Pointer(newTimingWheel(
-					tw.interval,
-					tw.wheelSize,
-					currentTime,
-					tw.queue,
-				)),
-			)
-			overflowWheel = atomic.LoadPointer(&tw.overflowWheel)
-		}
-		return (*TimingWheel)(overflowWheel).add(t)
-	}
-}
-
-// addOrRun inserts the timer t into the current timing wheel, or run the
-// timer's task if it has already expired.
-func (tw *TimingWheel) addOrRun(t *Timer) {
-	if !tw.add(t) {
-		// Already expired
-
-		// Like the standard time.AfterFunc (https://golang.org/pkg/time/#AfterFunc),
-		// always execute the timer's task in its own goroutine.
-		go t.task()
-	}
-}
-
-func (tw *TimingWheel) advanceClock(expiration int64) {
-	currentTime := atomic.LoadInt64(&tw.currentTime)
-	if expiration >= currentTime+tw.tick {
-		currentTime = truncate(expiration, tw.tick)
-		atomic.StoreInt64(&tw.currentTime, currentTime)
-
-		// Try to advance the clock of the overflow wheel if present
-		overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
-		if overflowWheel != nil {
-			(*TimingWheel)(overflowWheel).advanceClock(currentTime)
-		}
-	}
-}
-
-// Start starts the current timing wheel.
-func (tw *TimingWheel) Start() {
-	tw.waitGroup.Wrap(func() {
-		tw.queue.Poll(tw.exitC, func() int64 {
-			return timeToMs(time.Now().UTC())
-		})
-	})
-
-	tw.waitGroup.Wrap(func() {
-		for {
-			select {
-			case elem := <-tw.queue.C:
-				b := elem.(*bucket)
-				tw.advanceClock(b.Expiration())
-				b.Flush(tw.addOrRun)
-			case <-tw.exitC:
-				return
-			}
-		}
-	})
-}
-
-// Stop stops the current timing wheel.
-//
-// If there is any timer's task being running in its own goroutine, Stop does
-// not wait for the task to complete before returning. If the caller needs to
-// know whether the task is completed, it must coordinate with the task explicitly.
-func (tw *TimingWheel) Stop() {
-	close(tw.exitC)
-	tw.waitGroup.Wait()
-}
-
-// AfterFunc waits for the duration to elapse and then calls f in its own goroutine.
-// It returns a Timer that can be used to cancel the call using its Stop method.
-func (tw *TimingWheel) AfterFunc(d time.Duration, f func()) *Timer {
-	t := &Timer{
-		expiration: timeToMs(time.Now().UTC().Add(d)),
-		task:       f,
-	}
-	tw.addOrRun(t)
-	return t
-}
-
-// Scheduler determines the execution plan of a task.
-type Scheduler interface {
-	// Next returns the next execution time after the given (previous) time.
-	// It will return a zero time if no next time is scheduled.
-	//
-	// All times must be UTC.
-	Next(time.Time) time.Time
-}
-
-// ScheduleFunc calls f (in its own goroutine) according to the execution
-// plan scheduled by s. It returns a Timer that can be used to cancel the
-// call using its Stop method.
-//
-// If the caller want to terminate the execution plan halfway, it must
-// stop the timer and ensure that the timer is stopped actually, since in
-// the current implementation, there is a gap between the expiring and the
-// restarting of the timer. The wait time for ensuring is short since the
-// gap is very small.
-//
-// Internally, ScheduleFunc will ask the first execution time (by calling
-// s.Next()) initially, and create a timer if the execution time is non-zero.
-// Afterwards, it will ask the next execution time each time f is about to
-// be executed, and f will be called at the next execution time if the time
-// is non-zero.
-func (tw *TimingWheel) ScheduleFunc(s Scheduler, f func()) (t *Timer) {
-	expiration := s.Next(time.Now().UTC())
-	if expiration.IsZero() {
-		// No time is scheduled, return nil.
-		return
-	}
-
-	t = &Timer{
-		expiration: timeToMs(expiration),
-		task: func() {
-			// Schedule the task to execute at the next time if possible.
-			expiration := s.Next(msToTime(t.expiration))
-			if !expiration.IsZero() {
-				t.expiration = timeToMs(expiration)
-				tw.addOrRun(t)
-			}
-
-			// Actually execute the task.
-			f()
-		},
-	}
-	tw.addOrRun(t)
-
-	return
-}
@@ -1,26 +0,0 @@
-package kafkatimer
-
-import (
-	"fmt"
-	"testing"
-	"time"
-)
-
-func F()  {
-	fmt.Println("I'm timer...")
-}
-
-
-func TestTimingWheel_Start(t *testing.T) {
-	tw := NewTimingWheel(time.Millisecond, 20)
-	tw.Start()
-	defer tw.Stop()
-
-	exitC := make(chan time.Time, 1)
-	tw.AfterFunc(10 * time.Second, func() {
-		fmt.Println("The timer fires")
-		exitC <- time.Now().UTC()
-	})
-
-	fmt.Println(<-exitC)
-}
 \ No newline at end of file
@@ -1,38 +0,0 @@
-package kafkatimer
-
-import (
-	"sync"
-	"time"
-)
-
-// truncate returns the result of rounding x toward zero to a multiple of m.
-// If m <= 0, Truncate returns x unchanged.
-func truncate(x, m int64) int64 {
-	if m <= 0 {
-		return x
-	}
-	return x - x%m
-}
-
-// timeToMs returns an integer number, which represents t in milliseconds.
-func timeToMs(t time.Time) int64 {
-	return t.UnixNano() / int64(time.Millisecond)
-}
-
-// msToTime returns the UTC time corresponding to the given Unix time,
-// t milliseconds since January 1, 1970 UTC.
-func msToTime(t int64) time.Time {
-	return time.Unix(0, t*int64(time.Millisecond)).UTC()
-}
-
-type waitGroupWrapper struct {
-	sync.WaitGroup
-}
-
-func (w *waitGroupWrapper) Wrap(cb func()) {
-	w.Add(1)
-	go func() {
-		cb()
-		w.Done()
-	}()
-}
@@ -3,9 +3,12 @@ package net
 import (
 	"bufio"
 	"fmt"
+	"math"
 	"net"
 	"pro2d/src/common"
 	"pro2d/src/components/logger"
+	"pro2d/src/utils"
+	"sync/atomic"
 )
 type Head struct {
@@ -18,16 +21,20 @@ type Head struct {
 type Connection struct {
 	net.Conn
-	Id int
-	Server *Server
+	Id 				int
+	Server 			*Server
-	scanner *bufio.Scanner
-	writer *bufio.Writer
+	scanner 		*bufio.Scanner
+	writer 			*bufio.Writer
-	WBuffer chan []byte
-	RBuffer chan *MsgPkg
+	WBuffer 		chan []byte
+	RBuffer 		chan *MsgPkg
-	Quit chan *Connection
+	Quit 			chan *Connection
+
+	nextCheckTime   int64 //下一次检查的时间
+	lastHeartCheckTime int64 //最后收消息时间
+	heartTimeoutCount  int   //超时次数
 }
 type MsgPkg struct {
@@ -42,11 +49,13 @@ func NewConn(id int, conn net.Conn, s *Server) *Connection {
 		Conn: conn,
 		Server: s,
-		scanner: bufio.NewScanner(conn),
-		writer: bufio.NewWriter(conn),
-		WBuffer: make(chan []byte),
-		RBuffer: make(chan *MsgPkg),
-		Quit: make(chan *Connection),
+		scanner:            bufio.NewScanner(conn),
+		writer:             bufio.NewWriter(conn),
+		WBuffer:            make(chan []byte),
+		RBuffer:            make(chan *MsgPkg),
+		Quit:               make(chan *Connection),
+		lastHeartCheckTime: utils.Timex(),
+		heartTimeoutCount: 0,
 	}
 }
@@ -63,7 +72,6 @@ func (c *Connection) write()  {
 			logger.Error("write Flush fail err: " + err.Error())
 			return
 		}
-		logger.Debug("write :%s, n: %d", msg, n)
 	}
 }
@@ -83,6 +91,8 @@ func (c *Connection) read() {
 		//将请求消息发送给任务队列
 		c.Server.TaskQueue[workerID] <- req
+		atomic.StoreInt64(&c.lastHeartCheckTime, utils.Timex())
+
 		//备注，可以在当前协程处理当条请求(如下, 实现很简单，已经删除)，也可以丢到协程池里处理任务(如上)，还未对比效果。
 		//c.Server.OnRecv(req)
 	}
@@ -94,6 +104,41 @@ func (c *Connection) read() {
 	}
 }
+
+func (c *Connection) checkHeartBeat(now int64)  {
+	lastHeartCheckTime := atomic.LoadInt64(&c.lastHeartCheckTime)
+	logger.Debug("checkHeartBeat ID: %d, last: %d, now: %d", c.Id, lastHeartCheckTime, now)
+	if math.Abs(float64(lastHeartCheckTime - now)) > common.HEART_TIMER_INTERVAL {
+		c.heartTimeoutCount++
+		if c.heartTimeoutCount >= common.HEART_TIMEOUT_COUNT_MAX {
+			c.Quiting()
+			return
+		}
+		logger.Debug("timeout count: %d", c.heartTimeoutCount)
+	}else {
+		c.heartTimeoutCount = 0
+	}
+}
+
+func (c *Connection) update()  {
+	nextCheckTime := atomic.LoadInt64(&c.nextCheckTime)
+	now := utils.Timex()
+	if now >= nextCheckTime {
+		c.checkHeartBeat(now)
+		nextCheckTime = now + common.HEART_TIMER_INTERVAL
+		atomic.StoreInt64(&c.nextCheckTime, nextCheckTime)
+	}
+}
+//
+//func (c *Connection) SetLastHeartCheckTime()  {
+//	now := utils.Timex()
+//	lastHeartCheckTime := atomic.LoadInt64(&c.lastHeartCheckTime)
+//	if now - lastHeartCheckTime < common.HEART_TIMER_INTERVAL {
+//		logger.Debug("heart too quick")
+//	}
+//	atomic.StoreInt64(&c.lastHeartCheckTime, now)
+//}
+
 func (c *Connection) Start()  {
 	go c.write()
 	go c.read()
@@ -109,6 +154,7 @@ func (c *Connection) Stop()  {
 }
 func (c *Connection) Quiting()  {
+	logger.Debug("ID: %d close", c.Id)
 	c.Server.OnClose(c)
 }
@@ -11,12 +11,15 @@ import (
 	"pro2d/src/components/db"
 	"pro2d/src/components/etcd"
 	"pro2d/src/components/logger"
+	"pro2d/src/components/timewheel"
 	"pro2d/src/models"
 	"sync"
+	"time"
 )
 type ActionHandler func (msg *MsgPkg)  (int32, proto.Message)
 var ActionMap map[pb.ProtoCode]ActionHandler
+var TimeWheel *timewheel.TimeWheel
 type Server struct {
 	SConf      *conf.SConf
@@ -36,10 +39,6 @@ func NewServer(sConf *conf.SConf) *Server {
 	}
 }
-func (s *Server) StartTimer() {
-
-}
-
 //StartWorkerPool 启动worker工作池
 func (s *Server) StartWorkerPool() {
 	//遍历需要启动worker的数量，依此启动
@@ -96,6 +95,16 @@ func (s *Server) LoadPlugin() {
 	logger.Debug("load plugin success")
 }
+func (s *Server) handleTimeOut() {
+	s.Clients.Range(func(key, value interface{}) bool {
+		client := value.(*Connection)
+		client.update()
+		return true
+	})
+
+	TimeWheel.AfterFunc(1*time.Second, s.handleTimeOut)
+}
+
 func (s *Server)Start() error {
 	//mongo 初始化
 	db.MongoDatabase = db.MongoClient.Database(conf.GlobalConf.GameConf.DBName)
@@ -110,10 +119,10 @@ func (s *Server)Start() error {
 	s.EtcdClient.PutWithLeasePrefix(conf.GlobalConf.GameConf.Name, conf.GlobalConf.GameConf.ID, fmt.Sprintf("%s:%d", conf.GlobalConf.GameConf.IP, conf.GlobalConf.GameConf.Port), 5)
 	//初始化plugin
-	_, err = plugin.Open(conf.GlobalConf.GameConf.PluginPath)
-	if err != nil {
-		return err
-	}
+	//_, err = plugin.Open(conf.GlobalConf.GameConf.PluginPath)
+	//if err != nil {
+	//	return err
+	//}
 	port := fmt.Sprintf(":%d", s.SConf.Port)
 	l, err := net.Listen("tcp", port)
@@ -121,8 +130,15 @@ func (s *Server)Start() error {
 		return err
 	}
+	//启动协程池
 	s.StartWorkerPool()
+	//启动定时器
+	TimeWheel = timewheel.NewTimeWheel(common.TickMS * time.Millisecond, common.WheelSize)
+	TimeWheel.Start()
+	s.handleTimeOut()
+
+	//监听端口
 	logger.Debug("listen on %s\n", port)
 	id := 0
 	for {
@@ -139,5 +155,5 @@ func (s *Server)Start() error {
 }
 func (s *Server)Stop()  {
-}
-
+	TimeWheel.Stop()
+}
 \ No newline at end of file
@@ -1 +0,0 @@
-package skynettimer
-package kafkatimer
+package timewheel
 import (
 	"container/list"
@@ -6,97 +6,52 @@ import (
 	"sync/atomic"
 	"unsafe"
 )
-
-// Timer represents a single event. When the Timer expires, the given
-// task will be executed.
-type Timer struct {
-	expiration int64 // in milliseconds
-	task       func()
-
-	// The bucket that holds the list to which this timer's element belongs.
-	//
-	// NOTE: This field may be updated and read concurrently,
-	// through Timer.Stop() and Bucket.Flush().
-	b unsafe.Pointer // type: *bucket
-
-	// The timer's element.
-	element *list.Element
-}
-
-func (t *Timer) getBucket() *bucket {
-	return (*bucket)(atomic.LoadPointer(&t.b))
-}
-
-func (t *Timer) setBucket(b *bucket) {
-	atomic.StorePointer(&t.b, unsafe.Pointer(b))
-}
-
-// Stop prevents the Timer from firing. It returns true if the call
-// stops the timer, false if the timer has already expired or been stopped.
-//
-// If the timer t has already expired and the t.task has been started in its own
-// goroutine; Stop does not wait for t.task to complete before returning. If the caller
-// needs to know whether t.task is completed, it must coordinate with t.task explicitly.
-func (t *Timer) Stop() bool {
-	stopped := false
-	for b := t.getBucket(); b != nil; b = t.getBucket() {
-		// If b.Remove is called just after the timing wheel's goroutine has:
-		//     1. removed t from b (through b.Flush -> b.remove)
-		//     2. moved t from b to another bucket ab (through b.Flush -> b.remove and ab.Add)
-		// this may fail to remove t due to the change of t's bucket.
-		stopped = b.Remove(t)
-
-		// Thus, here we re-get t's possibly new bucket (nil for case 1, or ab (non-nil) for case 2),
-		// and retry until the bucket becomes nil, which indicates that t has finally been removed.
-	}
-	return stopped
-}
-
 type bucket struct {
-	// 64-bit atomic operations require 64-bit alignment, but 32-bit
-	// compilers do not ensure it. So we must keep the 64-bit field
-	// as the first field of the struct.
-	//
-	// For more explanations, see https://golang.org/pkg/sync/atomic/#pkg-note-BUG
-	// and https://go101.org/article/memory-layout.html.
+	//过期时间
 	expiration int64
-	mu     sync.Mutex
+	mu sync.Mutex
+	//相同过期时间的任务队列
 	timers *list.List
 }
-func newBucket() *bucket {
+func newBucket() *bucket  {
 	return &bucket{
-		timers:     list.New(),
 		expiration: -1,
+		mu:      sync.Mutex{},
+		timers:  list.New(),
 	}
 }
-func (b *bucket) Expiration() int64 {
-	return atomic.LoadInt64(&b.expiration)
-}
-
-func (b *bucket) SetExpiration(expiration int64) bool {
-	return atomic.SwapInt64(&b.expiration, expiration) != expiration
+func (b *bucket) SetExpiration(expiration int64)  {
+	atomic.AddInt64(&b.expiration, expiration)
 }
 func (b *bucket) Add(t *Timer) {
 	b.mu.Lock()
+	defer b.mu.Unlock()
 	e := b.timers.PushBack(t)
 	t.setBucket(b)
 	t.element = e
+}
+
+func (b *bucket) Flush(reinsert func(*Timer)) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	for e := b.timers.Front(); e != nil; {
+		next := e.Next()
+		t := e.Value.(*Timer)
+		b.remove(t)
-	b.mu.Unlock()
+		reinsert(t)
+		e = next
+	}
 }
 func (b *bucket) remove(t *Timer) bool {
 	if t.getBucket() != b {
-		// If remove is called from within t.Stop, and this happens just after the timing wheel's goroutine has:
-		//     1. removed t from b (through b.Flush -> b.remove)
-		//     2. moved t from b to another bucket ab (through b.Flush -> b.remove and ab.Add)
-		// then t.getBucket will return nil for case 1, or ab (non-nil) for case 2.
-		// In either case, the returned value does not equal to b.
 		return false
 	}
 	b.timers.Remove(t.element)
@@ -108,26 +63,23 @@ func (b *bucket) remove(t *Timer) bool {
 func (b *bucket) Remove(t *Timer) bool {
 	b.mu.Lock()
 	defer b.mu.Unlock()
+
 	return b.remove(t)
 }
-func (b *bucket) Flush(reinsert func(*Timer)) {
-	b.mu.Lock()
-	defer b.mu.Unlock()
-
-	for e := b.timers.Front(); e != nil; {
-		next := e.Next()
-
-		t := e.Value.(*Timer)
-		b.remove(t)
-		// Note that this operation will either execute the timer's task, or
-		// insert the timer into another bucket belonging to a lower-level wheel.
-		//
-		// In either case, no further lock operation will happen to b.mu.
-		reinsert(t)
+type Timer struct {
+	expiration int64
+	//要被执行的任务
+	task func()
-		e = next
-	}
+	b unsafe.Pointer
+	element *list.Element
+}
-	b.SetExpiration(-1)
+func (t *Timer) setBucket(b *bucket)  {
+	atomic.StorePointer(&t.b, unsafe.Pointer(b))
 }
+
+func (t *Timer) getBucket() *bucket {
+	return (*bucket)(atomic.LoadPointer(&t.b))
+}
 \ No newline at end of file
@@ -0,0 +1,165 @@
+package timewheel
+
+import (
+	"pro2d/src/common"
+	"pro2d/src/components/workpool"
+	"sync/atomic"
+	"time"
+	"unsafe"
+)
+
+type TimeWheel struct {
+	ticker    *time.Ticker
+	tickMs    int64 //一滴答的时间 1ms 可以自定义 我们这里选择使用1ms
+	wheelSize int64
+	startMs   int64 //开始时间 in millisecond
+	endMs     int64
+	wheelTime int64 //跑完一圈所需时间
+	level 	  int64 //层级
+
+	//时间刻度 列表
+	bucket []*bucket
+	currentTime int64 //当前时间 in millisecond
+	prevflowWheel unsafe.Pointer // type: *TimingWheel
+	overflowWheel unsafe.Pointer // type: *TimingWheel
+	exitC     chan struct{}
+
+	WorkPool *workpool.WorkPool
+}
+
+func NewTimeWheel(tick time.Duration, wheelSize int64) *TimeWheel {
+	//转化为毫秒
+	tickMs := int64(tick / time.Millisecond)
+	//如果小于零
+	if tickMs <=0 {
+		panic("tick must be greater than or equal to 1 ms")
+	}
+
+	startMs := time.Now().UnixMilli() //ms
+
+	workpool := workpool.NewWorkPool(common.WorkerPoolSize, common.MaxTaskPerWorker)
+	return newTimingWheel(tickMs, wheelSize, startMs, 0, nil, workpool)
+}
+
+func newTimingWheel(tick, wheelSize int64, start, level int64, prev *TimeWheel, pool *workpool.WorkPool) *TimeWheel {
+	buckets := make([]*bucket, wheelSize)
+	for i := range buckets {
+		buckets[i] = newBucket()
+	}
+
+	return &TimeWheel{
+		tickMs:      tick,
+		wheelSize:   wheelSize,
+		startMs:     start,
+		endMs:       wheelSize * tick + start,
+		wheelTime:   wheelSize * tick,
+		bucket:      buckets,
+		currentTime: truncate(start, tick),
+		exitC:       make(chan struct{}),
+		WorkPool: pool,
+
+		prevflowWheel: unsafe.Pointer(prev),
+		level: level,
+	}
+}
+
+func truncate(dst, m int64) int64 {
+	return dst - dst%m
+}
+
+func (tw *TimeWheel) add(t *Timer) bool {
+	currentTime := atomic.LoadInt64(&tw.currentTime)
+	if t.expiration < currentTime + tw.tickMs {
+		return false
+	}else if t.expiration < currentTime + tw.wheelTime {
+		virtualID := t.expiration / tw.tickMs  //需要多少滴答数
+		b := tw.bucket[virtualID%tw.wheelSize] //pos = 所需滴答数 % wheelSize
+		b.Add(t)
+
+		b.SetExpiration(virtualID * tw.tickMs)
+	}else {
+		overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
+		if overflowWheel == nil {
+			level := atomic.LoadInt64(&tw.level) + 1
+			atomic.CompareAndSwapPointer(
+				&tw.overflowWheel,
+				nil,
+				unsafe.Pointer(newTimingWheel(tw.wheelTime, tw.wheelSize, currentTime, level, tw , tw.WorkPool)),
+				)
+			overflowWheel = atomic.LoadPointer(&tw.overflowWheel)
+		}
+		//递归添加到下一级定时器中
+		(*TimeWheel)(overflowWheel).add(t)
+	}
+
+	return true
+}
+
+func (tw *TimeWheel) addOrRun(t *Timer) {
+	if !tw.add(t) {
+		workerID := t.expiration % tw.WorkPool.WorkerPoolSize
+		//将请求消息发送给任务队列
+		tw.WorkPool.TaskQueue[workerID] <- t.task
+	}
+}
+
+//拨动时钟
+func (tw *TimeWheel) advanceClock(expiration int64) {
+	level := atomic.LoadInt64(&tw.level)
+	currentTime := truncate(expiration, tw.tickMs)
+	atomic.StoreInt64(&tw.currentTime, currentTime)
+
+	if level == 0 {
+		virtualID := expiration / tw.tickMs    //需要多少滴答数
+		b := tw.bucket[virtualID%tw.wheelSize] //pos = 所需滴答数 % wheelSize
+		b.Flush(tw.addOrRun)
+	} else {
+		prevflowWheel := atomic.LoadPointer(&tw.prevflowWheel)
+		if prevflowWheel != nil {
+			virtualID := expiration / tw.tickMs    //需要多少滴答数
+			b := tw.bucket[virtualID%tw.wheelSize] //pos = 所需滴答数 % wheelSize
+			b.Flush((*TimeWheel)(prevflowWheel).addOrRun)
+		}
+	}
+
+	//如果基础的时钟指针转完了一圈，则递归拨动下一级时钟
+	if currentTime >= tw.endMs {
+		atomic.StoreInt64(&tw.startMs, currentTime)
+		atomic.StoreInt64(&tw.endMs, currentTime + tw.wheelTime)
+
+		overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
+		if overflowWheel != nil {
+			(*TimeWheel)(overflowWheel).advanceClock(currentTime)
+		}
+	}
+}
+
+
+func (tw *TimeWheel) AfterFunc(d time.Duration, f func()) *Timer {
+	t := &Timer{
+		expiration: time.Now().UTC().Add(d).UnixMilli(),
+		task:       f,
+	}
+	tw.addOrRun(t)
+	return t
+}
+
+func (tw *TimeWheel) Start()  {
+	tw.ticker = time.NewTicker(time.Duration(tw.tickMs) * time.Millisecond)
+	tw.WorkPool.StartWorkerPool()
+
+	go func() {
+		for {
+			select {
+			case t := <- tw.ticker.C:
+				tw.advanceClock(t.UnixMilli())
+			case <- tw.exitC:
+				return
+			}
+		}
+	}()
+}
+
+func (tw *TimeWheel) Stop()  {
+	tw.exitC <- struct{}{}
+}
 \ No newline at end of file
@@ -0,0 +1,30 @@
+package timewheel
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+var tw *TimeWheel
+
+func Add()  {
+	fmt.Println("ADD : 123456")
+	tw.AfterFunc(6*time.Second, Add)
+}
+
+func Add1()  {
+	fmt.Println("GET : 78901112")
+	tw.AfterFunc(9*time.Second, Add1)
+}
+
+func TestTimeWheel_AfterFunc(t *testing.T) {
+
+	tw = NewTimeWheel(time.Second, 5)
+	tw.Start()
+	defer tw.Stop()
+
+
+	Add()
+	Add1()
+	time.Sleep(time.Second * 200)
+}
@@ -0,0 +1,41 @@
+package workpool
+
+type Job func()
+
+type WorkPool struct {
+	WorkerPoolSize int64
+	MaxTaskPerWorker int64
+	TaskQueue []chan Job
+}
+
+func NewWorkPool(poolSize, maxTaskSize int64) *WorkPool {
+	return &WorkPool{
+		WorkerPoolSize: poolSize,
+		MaxTaskPerWorker: maxTaskSize,
+		TaskQueue: make([]chan Job, poolSize),
+	}
+}
+
+//StartOneWorker 启动一个Worker工作流程
+func (wp *WorkPool) StartOneWorker(workerID int, taskQueue chan Job) {
+	//不断的等待队列中的消息
+	for {
+		select {
+		//有消息则取出队列的Request，并执行绑定的业务方法
+		case job := <-taskQueue:
+			_ = workerID
+			job()
+		}
+	}
+}
+
+func (wp *WorkPool) StartWorkerPool() {
+	//遍历需要启动worker的数量，依此启动
+	for i := 0; i < int(wp.WorkerPoolSize); i++ {
+		//一个worker被启动
+		//给当前worker对应的任务队列开辟空间
+		wp.TaskQueue[i] = make(chan Job,  wp.MaxTaskPerWorker)
+		//启动当前Worker，阻塞的等待对应的任务队列是否有消息传递进来
+		go wp.StartOneWorker(i, wp.TaskQueue[i])
+	}
+}
 \ No newline at end of file
-package main
+package plugin
 import (
 	"github.com/golang/protobuf/proto"
@@ -10,6 +10,7 @@ import (
 )
 func HeartRpc(msg *net.MsgPkg) (int32, proto.Message)   {
+	//msg.Conn.SetLastHeartCheckTime()
 	return 0, nil
 }
-package main
+package plugin
 import (
 	"pro2d/protos/pb"
@@ -55,6 +55,6 @@ func Md5V(str string) string  {
 	return hex.EncodeToString(h.Sum(nil))
 }
-func Timex() int64  {
-	return time.Now().UnixMicro()
+func Timex() int64 {
+	return time.Now().Unix()
 }
 \ No newline at end of file
@@ -16,7 +16,7 @@ func main() {
 	head := &net2.Head{
 		Length:   0,
-		Cmd:      int32(pb.ProtoCode_LoginRpc),
+		Cmd:      int32(pb.ProtoCode_LoginReq),
 		ErrCode:  0,
 		PreField: 0,
 	}