liulei/jx-callback
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
邹宗楠
2023-05-30 09:06:09 +08:00
parent d626fcf12d
commit 4b9c783b9a
3 changed files with 345 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

335
business/jxstore/event/event_tcp.go
View File

@@ -16,6 +16,14 @@ import (
"unicode/utf8" "unicode/utf8"
) )
var Poll *Pool
func init() {
Poll = NewPool(100)
Poll.Start()
}
//入口 //入口
func ListenTcp() { func ListenTcp() {
l, err := net.Listen("tcp", ":8000") l, err := net.Listen("tcp", ":8000")
@@ -31,7 +39,7 @@ func ListenTcp() {
break break
} }
go func() { fn := func() {
defer func() { defer func() {
// 捕获异常 防止waitGroup阻塞 // 捕获异常 防止waitGroup阻塞
if err := recover(); err != nil { if err := recover(); err != nil {
@@ -45,7 +53,23 @@ func ListenTcp() {
t = nil t = nil
return return
} }
}() }
Poll.AddJob(fn)
//go func() {
// defer func() {
// // 捕获异常 防止waitGroup阻塞
// if err := recover(); err != nil {
// fmt.Println("recover err = ", err)
// return
// }
// }()
//
// t := NewTcpClient()
// if err := handleConn(c, t); err != nil {
// t = nil
// return
// }
//}()
} }
} }
@@ -185,28 +209,56 @@ func (t *TcpClient) HandleTcpMessages(printNo string) {
return return
} }
go func() { fn := func() {
for { for {
if t.TimeoutMap[printNo] == true { if t.TimeoutMap[printNo] == true {
timeNow := time.Now() if time.Now().Unix()%3 == 0 {
timeStart := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 0, 0, 0, 0, timeNow.Location()) timeNow := time.Now()
timeEnd := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 23, 59, 59, 0, timeNow.Location()) timeStart := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 0, 0, 0, 0, timeNow.Location())
prints, _ := dao.GetPrintMsgs(db, printNo, []int{printMsgWait}, timeStart.AddDate(0, 0, -1), timeEnd, offset, pageSize) timeEnd := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 23, 59, 59, 0, timeNow.Location())
for _, printMsg := range prints { prints, _ := dao.GetPrintMsgs(db, printNo, []int{printMsgWait}, timeStart.AddDate(0, 0, -1), timeEnd, offset, pageSize)
printMsg.Status = printMsgAlreadyLoad for _, printMsg := range prints {
//先避免重复读再插到channel？ printMsg.Status = printMsgAlreadyLoad
if _, err := dao.UpdateEntity(db, printMsg, "Status"); err == nil { //先避免重复读再插到channel？
if err = t.addMsgChan(printMsg); err != nil { if _, err := dao.UpdateEntity(db, printMsg, "Status"); err == nil {
globals.SugarLogger.Debugf("HandleTcpMessages addMsgChan Err: %v", err) if err = t.addMsgChan(printMsg); err != nil {
globals.SugarLogger.Debugf("HandleTcpMessages addMsgChan Err: %v", err)
}
} }
} }
} else {
continue
} }
} else { } else {
globals.SugarLogger.Debugf("HandleTcpMessages timeout") globals.SugarLogger.Debugf("HandleTcpMessages timeout")
return return
} }
} }
}() }
Poll.AddJob(fn)
//go func() {
// for {
// if t.TimeoutMap[printNo] == true {
// timeNow := time.Now()
// timeStart := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 0, 0, 0, 0, timeNow.Location())
// timeEnd := time.Date(timeNow.Year(), timeNow.Month(), timeNow.Day(), 23, 59, 59, 0, timeNow.Location())
// prints, _ := dao.GetPrintMsgs(db, printNo, []int{printMsgWait}, timeStart.AddDate(0, 0, -1), timeEnd, offset, pageSize)
// for _, printMsg := range prints {
// printMsg.Status = printMsgAlreadyLoad
// //先避免重复读再插到channel？
// if _, err := dao.UpdateEntity(db, printMsg, "Status"); err == nil {
// if err = t.addMsgChan(printMsg); err != nil {
// globals.SugarLogger.Debugf("HandleTcpMessages addMsgChan Err: %v", err)
// }
// }
// }
// } else {
// globals.SugarLogger.Debugf("HandleTcpMessages timeout")
// return
// }
// }
//}()
} }
func (t *TcpClient) readTimeoutMap(key string) bool { func (t *TcpClient) readTimeoutMap(key string) bool {
@@ -223,99 +275,192 @@ func (t *TcpClient) doPrint(key string) (err error) {
if !t.isExistMsg(key) { if !t.isExistMsg(key) {
return err return err
} }
go func() {
for { fn := func() {
if t.TimeoutMap[key] == true { if t.TimeoutMap[key] == true {
select { select {
case printMsg, ok := <-t.MsgMap[key]: case printMsg, ok := <-t.MsgMap[key]:
if !ok { if !ok {
globals.SugarLogger.Debugf("doPrint err !ok ...") globals.SugarLogger.Debugf("doPrint err !ok ...")
return return
} }
var ( var (
data []byte data []byte
c net.Conn c net.Conn
) )
if printMsg != nil { if printMsg != nil {
if err = checkPrintMsg(db, printMsg); err == nil { if err = checkPrintMsg(db, printMsg); err == nil {
status := t.getPrintStatus(printMsg.PrintNo) status := t.getPrintStatus(printMsg.PrintNo)
switch status { switch status {
//只有在线才打印内容 //只有在线才打印内容
case printerStatusOnline: case printerStatusOnline:
if c = t.getPrintConn(printMsg.PrintNo); c != nil { if c = t.getPrintConn(printMsg.PrintNo); c != nil {
data, err = buildMsg(printMsg) data, err = buildMsg(printMsg)
}
case printerStatusOffline:
err = fmt.Errorf("打印机离线！")
case printerStatusOnlineWithoutPaper:
err = fmt.Errorf("打印机缺纸！")
default:
err = fmt.Errorf("打印机状态未知！")
} }
case printerStatusOffline:
err = fmt.Errorf("打印机离线！")
case printerStatusOnlineWithoutPaper:
err = fmt.Errorf("打印机缺纸！")
default:
err = fmt.Errorf("打印机状态未知！")
} }
if err != nil { }
printMsg.Status = printMsgErr if err != nil {
printMsg.Comment = err.Error() printMsg.Status = printMsgErr
dao.UpdateEntity(db, printMsg, "Status", "Comment") printMsg.Comment = err.Error()
delete(t.TimeoutMap, key) dao.UpdateEntity(db, printMsg, "Status", "Comment")
} else { delete(t.TimeoutMap, key)
if c != nil { } else {
if _, err = c.Write(data); err != nil { if c != nil {
globals.SugarLogger.Debugf("handleTcpMessages err [%v]", err) if _, err = c.Write(data); err != nil {
//close(t.TimeoutMap[key]) globals.SugarLogger.Debugf("handleTcpMessages err [%v]", err)
delete(t.TimeoutMap, key) //close(t.TimeoutMap[key])
} else { delete(t.TimeoutMap, key)
//等待回调 } else {
dataStr := <-t.CallBackMap[key] //等待回调
if dataStr != "" { dataStr := <-t.CallBackMap[key]
a, b := getCallbackMsgInfo(dataStr) if dataStr != "" {
t.changePrintMsg(dataStr, a, b) a, b := getCallbackMsgInfo(dataStr)
// 查询打印机是否扣费,未扣费就扣费,已经扣费不做处理 t.changePrintMsg(dataStr, a, b)
have, err := dao.QueryOrderDeductionRecord(db, b, utils.Int64ToStr(a)) // 查询打印机是否扣费,未扣费就扣费,已经扣费不做处理
if err != nil && !have { have, err := dao.QueryOrderDeductionRecord(db, b, utils.Int64ToStr(a))
// 扣除打印机账号金额 if err != nil && !have {
if err = dao.DeductionPrintBalance(db, b); err != nil { // 扣除打印机账号金额
globals.SugarLogger.Debugf("扣除用户打印机金额错误 %s", err) if err = dao.DeductionPrintBalance(db, b); err != nil {
} else { globals.SugarLogger.Debugf("扣除用户打印机金额错误 %s", err)
// 添加打印记录(支出记录)
if err = dao.AddPrintRecord(db, &model.PrintBillRecord{
CreatedAt: time.Now(),
UpdatedAt: time.Now(),
PrintNo: b,
PayType: 2,
PayMoney: 1, // 固定支出一分钱
OrderId: utils.Int64ToStr(a),
UserId: "",
}); err != nil {
globals.SugarLogger.Debugf("添加打印机订单支付记录错误 %s", err)
}
}
} else { } else {
globals.SugarLogger.Debugf("查询打印机扣费记录错误 %s", err) // 添加打印记录(支出记录)
} if err = dao.AddPrintRecord(db, &model.PrintBillRecord{
//判断音频暂停？ CreatedAt: time.Now(),
//收到打印成功回调后，如果消息中有音频，需要等待一下,等上一个音频播完 UpdatedAt: time.Now(),
//暂停时间就暂时取的sound标签内内容长度/2 PrintNo: b,
if sounds := regexpSoundSpan.FindStringSubmatch(printMsg.Content); len(sounds) > 0 { PayType: 2,
sound := sounds[1] PayMoney: 1, // 固定支出一分钱
lenTime := time.Duration(utf8.RuneCountInString(sound)) * time.Second OrderId: utils.Int64ToStr(a),
time.Sleep(lenTime / 2) UserId: "",
}); err != nil {
globals.SugarLogger.Debugf("添加打印机订单支付记录错误 %s", err)
}
} }
} else {
globals.SugarLogger.Debugf("查询打印机扣费记录错误 %s", err)
}
//判断音频暂停？
//收到打印成功回调后，如果消息中有音频，需要等待一下,等上一个音频播完
//暂停时间就暂时取的sound标签内内容长度/2
if sounds := regexpSoundSpan.FindStringSubmatch(printMsg.Content); len(sounds) > 0 {
sound := sounds[1]
lenTime := time.Duration(utf8.RuneCountInString(sound)) * time.Second
time.Sleep(lenTime / 2)
} }
} }
} }
} }
} else {
globals.SugarLogger.Debugf("msgMap is nil")
} }
} else {
globals.SugarLogger.Debugf("msgMap is nil")
} }
} else {
globals.SugarLogger.Debugf("doPrint timeout")
return
} }
} else {
globals.SugarLogger.Debugf("doPrint timeout")
return
} }
}() }
Poll.AddJob(fn)
//go func() {
// for {
// if t.TimeoutMap[key] == true {
// select {
// case printMsg, ok := <-t.MsgMap[key]:
// if !ok {
// globals.SugarLogger.Debugf("doPrint err !ok ...")
// return
// }
// var (
// data []byte
// c net.Conn
// )
// if printMsg != nil {
// if err = checkPrintMsg(db, printMsg); err == nil {
// status := t.getPrintStatus(printMsg.PrintNo)
// switch status {
// //只有在线才打印内容
// case printerStatusOnline:
// if c = t.getPrintConn(printMsg.PrintNo); c != nil {
// data, err = buildMsg(printMsg)
// }
// case printerStatusOffline:
// err = fmt.Errorf("打印机离线！")
// case printerStatusOnlineWithoutPaper:
// err = fmt.Errorf("打印机缺纸！")
// default:
// err = fmt.Errorf("打印机状态未知！")
// }
// }
// if err != nil {
// printMsg.Status = printMsgErr
// printMsg.Comment = err.Error()
// dao.UpdateEntity(db, printMsg, "Status", "Comment")
// delete(t.TimeoutMap, key)
// } else {
// if c != nil {
// if _, err = c.Write(data); err != nil {
// globals.SugarLogger.Debugf("handleTcpMessages err [%v]", err)
// //close(t.TimeoutMap[key])
// delete(t.TimeoutMap, key)
// } else {
// //等待回调
// dataStr := <-t.CallBackMap[key]
// if dataStr != "" {
// a, b := getCallbackMsgInfo(dataStr)
// t.changePrintMsg(dataStr, a, b)
// // 查询打印机是否扣费,未扣费就扣费,已经扣费不做处理
// have, err := dao.QueryOrderDeductionRecord(db, b, utils.Int64ToStr(a))
// if err != nil && !have {
// // 扣除打印机账号金额
// if err = dao.DeductionPrintBalance(db, b); err != nil {
// globals.SugarLogger.Debugf("扣除用户打印机金额错误 %s", err)
// } else {
// // 添加打印记录(支出记录)
// if err = dao.AddPrintRecord(db, &model.PrintBillRecord{
// CreatedAt: time.Now(),
// UpdatedAt: time.Now(),
// PrintNo: b,
// PayType: 2,
// PayMoney: 1, // 固定支出一分钱
// OrderId: utils.Int64ToStr(a),
// UserId: "",
// }); err != nil {
// globals.SugarLogger.Debugf("添加打印机订单支付记录错误 %s", err)
// }
// }
// } else {
// globals.SugarLogger.Debugf("查询打印机扣费记录错误 %s", err)
// }
// //判断音频暂停？
// //收到打印成功回调后，如果消息中有音频，需要等待一下,等上一个音频播完
// //暂停时间就暂时取的sound标签内内容长度/2
// if sounds := regexpSoundSpan.FindStringSubmatch(printMsg.Content); len(sounds) > 0 {
// sound := sounds[1]
// lenTime := time.Duration(utf8.RuneCountInString(sound)) * time.Second
// time.Sleep(lenTime / 2)
// }
// }
// }
// }
// }
// } else {
// globals.SugarLogger.Debugf("msgMap is nil")
// }
// }
// } else {
// globals.SugarLogger.Debugf("doPrint timeout")
// return
// }
//
// }
//}()
return err return err
} }

22
business/jxstore/event/print_test.go
View File

@@ -8,6 +8,7 @@ import (
"strconv" "strconv"
"strings" "strings"
"testing" "testing"
"time"
"unicode/utf8" "unicode/utf8"
) )
@@ -82,3 +83,24 @@ func TestTen216(t *testing.T) {
fmt.Println(gg) fmt.Println(gg)
fmt.Println(kk) fmt.Println(kk)
} }
func TestNewPool(t *testing.T) {
var pool = NewPool(100)
pool.Start()
for i := 0; i < 10; i++ {
num := i
pool.AddJob(func() {
fmt.Printf("Worker %d: %d is odd? %v\n", num%5, num, num%2 == 1)
})
}
time.Sleep(5 * time.Second)
for i := 0; i < 10; i++ {
pool.AddJob(func() {
fmt.Println("刘磊")
})
}
pool.Wait()
pool.Stop()
}

130
business/jxstore/event/timing_task.go
View File

@@ -1,49 +1,85 @@
package event package event
// import (
//import ( "sync"
// "fmt" )
// "sync"
//) type Job func()
//
//type SimplePool struct { type Worker struct {
// wg sync.WaitGroup id int
// work chan func() //任务队列 jobChannel chan Job
//} quit chan bool
// }
//func NewSimplePoll(workers int) *SimplePool {
// p := &SimplePool{ type Pool struct {
// wg: sync.WaitGroup{}, workers []*Worker
// work: make(chan func()), jobChannel chan Job
// } wg sync.WaitGroup
// p.wg.Add(workers) }
// //根据指定的并发量去读取管道并执行
// for i := 0; i < workers; i++ { func NewWorker(id int, jobChannel chan Job) *Worker {
// go func() { return &Worker{
// defer func() { id: id,
// // 捕获异常 防止waitGroup阻塞 jobChannel: jobChannel,
// if err := recover(); err != nil { quit: make(chan bool),
// fmt.Println(err) }
// p.wg.Done() }
// }
// }() func (w *Worker) Start() {
// // 从workChannel中取出任务执行 go func() {
// for fn := range p.work { for {
// fn() select {
// } case job := <-w.jobChannel:
// p.wg.Done() job()
// }() case <-w.quit:
// } return
// return p }
//} }
// }()
//// 添加任务 }
//func (p *SimplePool) Add(fn func()) {
// p.work <- fn func (w *Worker) Stop() {
//} go func() {
// w.quit <- true
//// 执行 }()
//func (p *SimplePool) Run() { }
// close(p.work)
// p.wg.Wait() func NewPool(numWorkers int) *Pool {
//} jobChannel := make(chan Job)
pool := &Pool{
workers: make([]*Worker, numWorkers),
jobChannel: jobChannel,
}
for i := 0; i < numWorkers; i++ {
worker := NewWorker(i, jobChannel)
pool.workers[i] = worker
}
return pool
}
func (p *Pool) Start() {
for _, worker := range p.workers {
worker.Start()
}
}
func (p *Pool) Stop() {
for _, worker := range p.workers {
worker.Stop()
}
}
func (p *Pool) AddJob(job Job) {
p.wg.Add(1)
p.jobChannel <- func() {
job()
p.wg.Done()
}
}
func (p *Pool) Wait() {
p.wg.Wait()
}