c++ - c++中异步线程的实现

Question

我无法在 C++11 中找到异步线程的正确用法。我想做的是我想要 spwan 线程并且每个线程将同时运行而不像 thread.join() 那样互相等待，这使得其他线程等待当前线程完成。那么，C++中是否有任何库可以使线程并行运行同时完成它们的工作而不必等待另一个完成。实际上我想要的是我想要同时运行每个线程，这样它们就不会等待另一个线程完成，并且它的功能是同时执行的，而不必等待其他线程完成。 谢谢， 库沙尔

编辑: 编辑::我在下面发布代码

#include <signal.h>
#include <thread>
#include <algorithm>
#include <cstring>
#include <csignal>
#include "paho_client.h"
using namespace std;
vector<string>    topic_container{"rpi2/temp","sense                          /bannana","sense/util","mqtt/temp","sense/temp","sense/pine","sense/fortis/udap"};
 vector<paho_client> publisher;
 vector<paho_client> subscriber;
 int finish_thread=1;
 void Onfinish(int signum){
 finish_thread=0;
 exit(EXIT_FAILURE);
 }

 int main(int argc, char** argv) {
 signal(SIGINT, Onfinish);
 int topic_index;
 if(argc<3){
    cout<<"the format of starting commandline argument is"<<endl;              
    exit(1);
    }

    while(finish_thread!=0){
    //paho_client::get_library_handle();
    if(strcmp(argv[1],"create_publisher")){
        for(topic_index=0;topic_index<atoi(argv[2]);topic_index++){
            thread pub_th;
            pub_th = thread([ = ]() {
                paho_client client("publisher", "192.168.0.102", "9876",
                                   topic_container[topic_index].c_str());
                client.paho_connect_withpub();
              publisher.push_back(client);
            });
         pub_th.join();
        }
        vector<paho_client>::iterator it;
        int publisher_traverse=0;
        for(it=publisher.begin();it<publisher.end();publisher_traverse++){
           publisher[publisher_traverse].increment_count();
          publisher[publisher_traverse].get_count();
       }

   }
  }
 return 0;
}

在使用 async with future 之后，我得到了与上面相同的行为，请指出我哪里出错了

 #include <signal.h>
 #include <thread>
 #include <algorithm>
 #include <cstring>
#include <csignal>
#include <future>
#include "paho_client.h"
using namespace std;
vector<string> topic_container{"rpi2/temp","sense/apple","sense/bannana","sense/util","mqtt/temp","sense/temp","sense/pine","sense/fortis/udap"};
vector<paho_client> publisher;
vector<paho_client> subscriber;
int finish_thread=1;
void Onfinish(int signum){
finish_thread=0;
exit(EXIT_FAILURE);
}
int accumulate_block_worker_ret(int topic_index) {
//int topic_index=0;
paho_client client("publisher", "192.168.0.102", "9876",
                   topic_container[topic_index].c_str());
client.paho_connect_withpub();
publisher.push_back(client);
client.increment_count();
return client.get_count();
 }


    int main(int argc, char** argv) {
    signal(SIGINT, Onfinish);

    if(argc<3){
    cout<<"the format of starting commandline argument is . /paho_client_emulate <create_publisher><count of publisher client to spawn>"  <<endl;
    exit(1);
   }

     while(finish_thread!=0){
//   paho_client::get_library_handle();
     int topic_index;
      if(strcmp(argv[1],"create_publisher")){
     for(topic_index=0;topic_index<atoi(argv[2]);topic_index++){
    //  thread pub_th;
    // pub_th = thread([ = ]() {
       future<int> f =  async(std::launch::async,accumulate_block_worker_ret,topic_index);
    //      });
    //  pub_th.join();
        cout<<"the returned value from future is"<<f.get()<<endl;
       }

    vector<paho_client>::iterator it;
    int publisher_traverse=0;
    for(it=publisher.begin();it<=publisher.end();publisher_traverse++){
        cout<<"came here"<<endl;
        publisher[publisher_traverse].increment_count();
        publisher[publisher_traverse].get_count();
     }

     }
     }
     return 0;
    }

Answer 1

i want to launch all the publisher clients first (as threads) and later publish messages from each threads

pub_th.join() 被错误地放置在线程启动的循环中，因此在启动下一个线程之前等待每个线程的终止。要让线程并行运行，只需将 .join() 移到该循环之外。当然，要访问循环体之后的线程，它们必须存储在某个地方，例如。 G。在 vector 中 - 为此，将第一个 for 循环更改为

        vector <thread> pub_threads;
        for (topic_index=0; topic_index<atoi(argv[2]); topic_index++)
        {
            pub_threads.push_back(thread([ = ]() { /* whatever */ }));
        }

稍后完成:

        for (auto &th: pub_threads) th.join();

Actually i am running infinite while inside every instance of paho_client so the first thread is not completed … that thread is run continously

当然，如果从未完成，那么 .join() 就没有意义了。