requestthreadpool.cpp 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150
  1. #include "requestthreadpool.h"
  2. #include <QTimer>
  3. #include <QDebug>
  4. namespace TC {
  5. RequestThreadPool* RequestThreadPool::s_instance = nullptr;
  6. RequestThreadPool* RequestThreadPool::instance()
  7. {
  8. if (!s_instance) {
  9. s_instance = new RequestThreadPool();
  10. }
  11. return s_instance;
  12. }
  13. RequestThreadPool::RequestThreadPool(QObject* parent)
  14. : QObject(parent)
  15. {
  16. // 默认设置为4个线程
  17. m_threadPool.setMaxThreadCount(4);
  18. }
  19. RequestThreadPool::~RequestThreadPool()
  20. {
  21. m_threadPool.waitForDone();
  22. }
  23. void RequestThreadPool::setMaxThreadCount(int count)
  24. {
  25. m_threadPool.setMaxThreadCount(count);
  26. }
  27. int RequestThreadPool::activeThreadCount() const
  28. {
  29. return m_threadPool.activeThreadCount();
  30. }
  31. void RequestThreadPool::addTask(std::shared_ptr<RequestTask> task)
  32. {
  33. if (!task) {
  34. return;
  35. }
  36. // 连接信号槽,传递任务指针作为参数
  37. connect(task.get(), &RequestTask::taskCompleted, this, &RequestThreadPool::onTaskCompleted);
  38. connect(task.get(), &RequestTask::taskFailed, this, [this, task]() {
  39. this->onTaskFailed(task.get());
  40. });
  41. QMutexLocker locker(&m_mutex);
  42. m_taskQueue.enqueue(task);
  43. // 如果线程池有空闲线程,则立即处理任务
  44. if (m_threadPool.activeThreadCount() < m_threadPool.maxThreadCount()) {
  45. std::shared_ptr<RequestTask> nextTask = nullptr;
  46. if (!m_taskQueue.isEmpty()) {
  47. nextTask = m_taskQueue.dequeue();
  48. m_activeTaskCount++;
  49. }
  50. // 在锁释放后启动任务,避免死锁
  51. if (nextTask) {
  52. m_threadPool.start(nextTask.get());
  53. }
  54. }
  55. }
  56. void RequestThreadPool::processNextTask()
  57. {
  58. std::shared_ptr<RequestTask> nextTask = nullptr;
  59. {
  60. QMutexLocker locker(&m_mutex);
  61. if (!m_taskQueue.isEmpty()) {
  62. nextTask = m_taskQueue.dequeue();
  63. m_activeTaskCount++;
  64. }
  65. }
  66. // 在锁释放后启动任务,避免死锁
  67. if (nextTask) {
  68. m_threadPool.start(nextTask.get());
  69. }
  70. }
  71. void RequestThreadPool::onTaskCompleted(bool success)
  72. {
  73. {
  74. QMutexLocker locker(&m_mutex);
  75. m_activeTaskCount--;
  76. }
  77. // 在锁释放后处理下一个任务
  78. processNextTask();
  79. }
  80. // 修改方法签名,直接接收任务指针作为参数
  81. void RequestThreadPool::onTaskFailed(RequestTask* task)
  82. {
  83. if (!task) {
  84. return;
  85. }
  86. bool canRetry = false;
  87. int retryCount = 0;
  88. {
  89. QMutexLocker locker(&m_mutex);
  90. m_activeTaskCount--;
  91. // 在锁内检查是否可以重试,但不执行重试操作
  92. canRetry = task->canRetry();
  93. if (canRetry) {
  94. task->incrementRetryCount();
  95. retryCount = task->retryCount();
  96. }
  97. }
  98. if (canRetry) {
  99. qDebug() << "Task failed, retrying..." << retryCount << "of" << task->maxRetries();
  100. // 延迟一段时间后重试 (500ms * 重试次数)
  101. QTimer::singleShot(500 * retryCount, [this, task]() {
  102. bool shouldRetry = task->retry();
  103. if (shouldRetry) {
  104. // 重新加入队列
  105. bool startTask = false;
  106. {
  107. QMutexLocker locker(&m_mutex);
  108. m_activeTaskCount++;
  109. startTask = true;
  110. }
  111. if (startTask) {
  112. m_threadPool.start(task);
  113. }
  114. } else {
  115. // 重试失败,处理下一个任务
  116. processNextTask();
  117. }
  118. });
  119. } else {
  120. // 达到最大重试次数,处理下一个任务
  121. processNextTask();
  122. }
  123. }
  124. } // namespace TC