定时推送之前失败过的消息
()
| 73 | |
| 74 | // 定时推送之前失败过的消息 |
| 75 | func (m *messageDB) repush() { |
| 76 | for ; ; time.Sleep(repushTime) { |
| 77 | var ids []int64 |
| 78 | now := time.Now() |
| 79 | |
| 80 | m.Lock() |
| 81 | for id, tm := range m.retry { |
| 82 | if now.Before(tm) { |
| 83 | ids = append(ids, id) |
| 84 | } |
| 85 | } |
| 86 | m.Unlock() |
| 87 | |
| 88 | for _, id := range ids { |
| 89 | t := time.Now().Unix() |
| 90 | log.Debugf("debug 1, time:%v", t) |
| 91 | |
| 92 | msgs, err := m.get(id) |
| 93 | if err != nil { |
| 94 | log.Errorf("get message:%d error:%s", id, errors.ErrorStack(err)) |
| 95 | m.addRetry(id) |
| 96 | continue |
| 97 | } |
| 98 | |
| 99 | log.Debugf("sender send msg:%v", msgs[0]) |
| 100 | if err = m.sender.send(*msgs[0]); err != nil { |
| 101 | if errors.Cause(err) != ErrInvalidSender { |
| 102 | log.Errorf("push message:%+v error:%s", msgs[0], errors.ErrorStack(err)) |
| 103 | //TODO 这块应该考虑使用LRU算法，把不在线的放在最后去重试，优先推送在线的消息，离线用户多的话可以增加 |
| 104 | //消息推送的效率??? |
| 105 | m.addRetry(id) |
| 106 | continue |
| 107 | } |
| 108 | } |
| 109 | //推送成功删除消息 |
| 110 | m.delRetry(id) |
| 111 | log.Debugf("debug 2, time:%v", time.Now().Unix()-t) |
| 112 | log.Debugf("remove retry message:%d", id) |
| 113 | } |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | func (m *messageDB) addRetry(id int64) { |
| 118 | log.Debugf("id:%v", id) |