2019-01-03

C++11 多线程编程

C++11 多线程编程
C++ Concurrency in Action

线程是用来运行程序的，而进程不是

class myclass{
    int &myint;
    myclass(int &m_i)
        :myint(m_i){}
    void operator()()
    {
        cout<<"sub thread operator()"<<endl;
    }
};
int num=7;
myclass mc(num);
thread mythread(mc);
mythread.detach();
cout<<"main  thread is going to over!"<<endl;
//当主线程结束时，将销毁自动变量num，子线程仍然在使用num
//调用myclass的拷贝构造函数，复制到mythread该线程

3 线程传参，detach坑，成员函数作为线程函数

1.传递临时对象作为线程参数

void myprint(const int&i,char* pmybuff)
{}
int myvar=1;
int &myvary=myvar;
char mybuf[]="test";

thread mytobj(myprint,myvar,mybuf);
mytobj.join();
////
mytobj.detach();
//myprint 中 myvar 复制给i,&i != &myvar,实际值传递 detach时i使用没问题
//pmybuff= mybuf
//指针在detach子线程时，有问题
void myprint(const int&i,const string& pmybuff)// 依靠自动类型转换,生成临时变量 detach时没问题
//mybuf 有可能在main执行完被回收后才转为string
    
thread mytobj(myprint,myvar,string(mybuf))；///！！！！！！！！！！！！！！
    //传递一个临时对象给myprint
//在创建线程的同时，创建临时对象是可行的。，在主线程结束之前被构造出来

只要通过构造临时A对象作为参数传递给线程，那么就一定能够在主线程执行完毕前将线程函数的第二个参数构造出来。
若直接Int简单类型，直接值传递
如果class对象，避免隐式类型转换，在创建线程的同时，创建临时对象，线程函数参数用引用，否则，会再拷贝构造一次对象
局部变量失效导致线程对内存的非法引用问题

2.临时对象作为线程参数继续讲

线程int id this_thread::get_id();

1 2	//void myprint(1,"123") 隐式转换在子线程中转换 //void myprint(1,string("123"))临时对象在主线程中被构造，构造函数，拷贝构造都在主线程

3.传递类对象、智能指针作为线程参数

thread mytobj(myprint2,myobj)尽管引用，调用的是拷贝构造函数，所以线程改变参数时，无法影响main函数
myprint2(const A& pybuff) //传递不是引用时，必须加const,否则报错
std::ref() 传递引用参数，可以去掉const，不会调用拷贝构造函数

void myprint2(unique_ptr<int> pzn)
{
  unique_ptr<int> myp(new int(100));
  thread mytobj(myprint2,std::move(myp));
  mytobj.join()//不能detach，new int(100) 释放，而线程仍指向该内存 
}

### 4.用成员函数指针作为线程参数


class A{
    void thread_work(int num)
    {
        
    }
};
A myobj(10);
thread mytobj(&A::thread_work,myobj,15)；// 调拷贝构造myobj
thread mytobj(&A::thread_work,ref(myobj),15)；====thread mytobj(&A::thread_work,&myobj,15)；    //两种模式相同，不调用拷贝构造，detach不安全
mytobj.join();

4 创建多个线程、数据共享问题分析

1.创建和等待多个线程

void myprint(int num)
{
    
}
vector<thread> mythreads;
for(int i=0;i<10;i++)
    mythreads.push(thread(myprint,i));//创建同时，已在执行
for(auto iter=mythreads.begin();iter!=mythreads.end();++iter)
    iter->join();/
cout<<"end"<<endl;//所有子线程结束后，主线程才结束

2.数据共享问题分析

只读数据没问题
有读有写

3.共享数据的保护案例代码

一个线程收集玩家命令，写入到队列中
从队列中取出命令

list 频繁按顺序插入删除数据 vector 随机插入删除

class A{
    public:
    int inMsgRecvQueue()
    {
        for(int i=0;i<1000000;i++)
            msgRecvQueue.push_back(i);
    }
    void outMsgRecvQueue()
    {
        for(int i=0;i<10000000;i++)
        {
            if(!msgRecvQueue.empty())
            {
                int cmd=msgRecvQueue.front();
                msgRecvQueue.pop_front();
            }
            else
                cout<<"list empty"<<endl;
        }
    }
    private:
    	list<int> msgRecvQueue;
};

void multithread_read_write_test()
{
    A myobj;
    thread myoutobj(&A::outMsgRecvQueue,&myobj);
    thread myinobj(&A::inMsgRecvQueue,&myobj);


    myinobj.join();
     myoutobj.join();
}

5.互斥量概念、用法、死锁

互斥量基本概念

一个类对象，多个线程可尝试Lock成员函数加锁，只有一个可以锁成功，标志时Lock函数返回。如果没有锁成功，会卡在lock这里，并不断尝试，原则是只保护需要保护的数据
if 多个分支一个lock对应一个unlock

#ifndef MUTEX_DEADLOCK_HPP
#define MUTEX_DEADLOCK_HPP

#include <list>
#include <iostream>
#include <thread>
#include <mutex>
using namespace std;
namespace  mutex_deadlock{


class A{
private:
    list<int> msgRecvQueue;
    mutex mymutex;
public:
    void inMsgRecvQueue()
    {
        for(int i=0;i<1000;i++)
        {
            cout<<"insert a number"<<endl;
            mymutex.lock();
            msgRecvQueue.push_back(i);
            mymutex.unlock();
        }
    }
    bool outMsgProc(int& command)
    {
          mymutex.lock();
        if(!msgRecvQueue.empty())
        {
            command=msgRecvQueue.front();
            msgRecvQueue.pop_front();
            mymutex.unlock();
            return true;
        }
        mymutex.unlock();
        return false;
    }
    void outMsgRecvQueue()
    {
        for(int i=0;i<1000;i++)
       {
            int cmd=0;
            bool result=outMsgProc(cmd);
           if(result)
           {
                cout<<"pop a number:: "<<cmd<<endl;
           }
           else
               cout<<"list empty..."<<endl;
       }
        return;
    }
};

void mutex_deadlock_test()
{
    A myobj;
    thread myoutobj(&A::outMsgRecvQueue,&myobj);
    thread myinobj(&A::inMsgRecvQueue,&myobj);


    myinobj.join();
    myoutobj.join();
}

}
#endif // MUTEX_DEADLOCK_HPP

lock_guard 的类模板，自动unlock,替代lock和unlock,在构造函数调用lock，析构函数一般在return调用unlock，或者加一个作用域标识符{}

1 2	mymutex.lock();// lock_guard<mutex> lguard(my_mutex);

互斥量用法

死锁

一个互斥量一把锁，死锁问题至少两把锁，至少两个互斥量，
线程A，先锁lockA, 成功，在锁lockB时，线程切换
线程B，先锁lockB，成功，等待锁lockA，线程A等待锁lockB

//threadA：
mutex1.lock();
mutex2.lock();

mutex2.unlock();
mutex1.unlock();

//threadB:
mutex2.lock();
mutex1.lock();

mutex2.unlock();
mutex1.unlock();

deadlock解决
- 保证互斥量上锁顺序一致，lock_guard上锁顺序一致
lock 函数模板
- 一次至少锁两个互斥量。同一时刻锁祝多个互斥量，不存在上锁顺序不一致时导致的死锁。
- 如果有一个没有锁住，就释放已经锁住的锁，否则全部锁住才返回。要么都锁住，要么都不锁住，
  1
  2
  3
  lock(mutex1,mutex2);
  mutex1.unlock();
  mutex2.unlock();

lock_guard

1
2
3

lock(mutex1,mutex2);
lock_guard<mutex> guard1(mutex1,adopt_lock);//不上锁
lock_guard<mutex> guard2(mutex2,adopt_lock);

adopt_lock 标识互斥量已经上锁

6.unique_lock

unique_lock取代lock_guard

类模板，一般lock_guard足够，lock_guard 构造加锁，析构解锁
unique_lock 效率差，内存多，灵活性高

unique_lock第二个参数

adopt_lock 标记已上锁（所以必须提前lock,否则异常），无需再上锁，假设调用方线程已经拥有了互斥的所有权
1
2
mymutex1.lock();
unique_lock<mutex> guard1(mymutex1,adopt_lock);

1 2	std::chrono::millseconds dura(20000);//20s this_thread.sleep_for(dura);

try_to_lock 尝试用mutex的lock锁Mutex，如果没有成功，也会立即返回。前提是,不能上锁

unique_lock<mutex> guard1(mymutex1,try_to_lock);
if(guard1.owns_lock())//如果拿到锁
{
    
}
else
{
    
}

defer_lock 前提不能lock，否则异常，初始化一个不加锁的mutex

unique_lock 成员函数

lock

unqiue_lock<mutex> guard1(mymutex1,defer_lock);//将mymutex1 绑定guard1
guard1.lock(); //不用担心unlock

// 处理共享代码
guard2.unlock()
//处理 非 共享代码
guard1.lock()

unlock
- 析构时就不会unlock,lock锁住的代码段越少，效率也高。锁的粒度–锁住的代码多少
- 锁力度细，锁住的代码少，效率高
try_lock
- 尝试加锁，如果拿到锁，返回true,否则false,不阻塞
release
- 返回所管理的mutex对象指针，并返回所有权，unique和mutex不再有联系
- 如果原来Mutex已加锁，就要负责自己unlock
- 1
  2
  3
  4
  mutex *ptx=guard1.release();
  //之后需要自己解锁mymutex
  //.....
  ptx->unlock();

unique_lock所有权传递

unqiue_lock<mutex> guard1(mymutex1）;
//guard1拥有mymutex1的所有权   
unqiue_lock<mutex> guard2(std::move(mymutex1)）;
                          
                          
unique_lock<mutex> trn_unique_lock()
{
   unique_lock<mutex> tmpguard(mymutex1);
   return tmpguard;//从函数返回一个局部的Unique_lock对象可以，移动构造函数
}
unqiue_lock<mutex> guard3= trn_unique_lock()；//从函数返回一个局部的Unique_lock对象可以，移动构造函数

所有权可以转移，不可复制
move
返回一个局部对象unique_lock

7.单例设计模式共享数据分析、解决，call_once

1设计模式

2 单例设计模式

一个类的实例只有一个

class mycas
{
private:
    mycas(){}
 private:
    static mycas* m_instance;
    static int num;
public:
    static mycas* getinstance()
    {
        if(m_instance==NULL)
        {
            m_instance=new mycas();
            static cgarrecycle cl;
        }
        return m_instance;
    }
    class cgarrecycle// 类中类，用来释放对象
    {
    public:
        ~cgarrecycle()
        {
            if(mycas::m_instance)
            {
                delete mycas::m_instance;
                mycas::m_instance=NULL;
            }
        }
    };

};

3 单例设计模式共享数据问题分析、解决

建议再主线程创建单例，（在创建子线程之前），之后创建子线程。

问题：

需要再自己创建的线程（非主线程）创建mycas实例，这种线程可能不止一个。
getinstance() 需要互斥,否则有可能创建多个实例

static mycas* getinstance()
{
    unique_lock<mutex> mymutex(resource_mutex);//可以 但 效率低
    if(m_instance==NULL)
    {
        m_instance=new mycas();
        static cgarrecycle cl;
    }
    return m_instance;
}
// if(m_instance!=NULL) 表示m_instance已经被new
// if(m_instance==NULL) 不表示m_instance一定没被new过，不加mutex
// 
static mycas* getinstance2()
{
    if(m_instance==NULL)// 双重锁定
    {
        unique_lock<mutex> mymutex(resource_mutex);//可以 但 效率低
        if(m_instance==NULL)
        {
            m_instance=new mycas();
            static cgarrecycle cl;
        }
     }
    return m_instance;
}

////////////////////////////////////
std::once_flat g_flag;
static void createinstance()
{
      m_instance=new mycas();
            static cgarrecycle cl;
}
static mycas* getinstance3()
{
    //两个线程同时执行到这里，一个线程须等待另一个线程执行完毕createinstance后，才
//阻塞
    std::call_once(g_flag,createinstance);
    return m_instance;
}
//////////////////////////////////
void mythread()
{
    cout<<"mythrad "<<endl;
    mycas *pa=mycas::getinstance();
    return;
}

void singleton_test()
{
    //mycas *pa=mycas::getinstance();

    thread thread1(mythread);
    thread thread2(mythread);
    
}

4 std::call_once()

第二个参数为函数名A
保证函数A只被调用一次
具备互斥量能力，比mutex更高效
需要与一个标记使用，std::once_flag一个结构体，通过结构体决定A()是否执行
调用call_once成功，就把这个标记设置为以调用状态，下次调用call_once A（）不会执行

static once_flag g_flag;   
static void createinstance()//只被调用一次
    {
        cout<<"createinstance  once"<<endl;
        m_instance=new mycas();
        static cgarrecycle cl;
    }

static mycas* getinstance2()
    {
       cout<<"getinstance2"<<endl;
       call_once(g_flag,createinstance);//两个线程同时执行到这里，一个线程须等待另一个线程执行完毕createinstance后，才
       return m_instance;
    }

call_once 用在成员函数时
this，作为call_once 第二个参数成员函数的参数

class X
{ 
    private:
connection_info connection_details;
connection_handle connection;
std::once_flag connection_init_flag;
void open_connection()
{
	connection=connection_manager.open(connection_details);
}
public:
X(connection_info const& connection_details_):
connection_details(connection_details_)
{}
void send_data(data_packet const& data) // 1
{
    std::call_once(connection_init_flag,&X::open_connection,this);
    // 2
    connection.send_data(data);
} 
    data_packet receive_data() // 3
  {
    std::call_once(connection_init_flag,&X::open_connection,this);
    // 2
    return connection.receive_data();
  }
}

8 condition_variable wait notify_one notify_all

1 条件变量condition_variable wait() norify_one()

线程A等待一个条件满足，从等待代码出，继续执行，阻塞
线程B往消息队列中扔消息（数据）
condition_variable 一个和条件相关的类，等待一个条件满足，需要和互斥量配合工作
用时，需要生成对象

2 条件变量导致的问题

#ifndef CONDITION_VARIABLE8_HPP
#define CONDITION_VARIABLE8_HPP


#include <list>
#include <iostream>
#include <thread>
#include  <mutex>

using namespace std;

namespace condition_variable_test
{
 class A{
private:
    list<int> msgRecvQueue;
    mutex mymutex1;
public:
    void inMsgRecvQueue()
    {
        for(int i=0;i<100;i++)
        {
            unique_lock<mutex> ulock(mymutex1);
            msgRecvQueue.push_back(i);
            cout<<"send a msg: "<<i<<endl;
        }
    }
    bool outMsgProc(int& command)
    {
        // 双重锁定
        if(!msgRecvQueue.empty()){
            
             unique_lock<mutex> ulock2(mymutex1);
             if(!msgRecvQueue.empty())
             {
                 command=msgRecvQueue.front();
                 msgRecvQueue.pop_front();
                 return true;
             }
        }
         else
             return false;
    }
    void outMsgRecvQueue()
    {
        int cmd;

            for(int i=0;i<100;i++)
           {
               if(outMsgProc(cmd))
               {
                   cout<<"receive a msg "<<cmd<<endl;
               }
               else
                   cout<<"list empty"<<endl;
           }
    }
};

    void condition_variable_test()
    {
        A myobj;
        thread myoutobj(&A::outMsgRecvQueue,&myobj);
        thread myinobj(&A::inMsgRecvQueue,&myobj);


        myinobj.join();
        myoutobj.join();
    }
}
#endif // CONDITION_VARIABLE8_HPP

condition_variable.wait
- 如果第二个lambda表达式返回值false,wait 解锁互斥量，并阻塞到本行，一直到其他某个线程norify_One()成员函数
- 返回true，wait直接返回
- 如果没有第二个参数，就和第二个参数返回false效果一样
  - wait 被notify_one 唤醒后，wait 不断尝试获得锁，加锁如果获取不到，阻塞，如果获取到，就继续
  - 唤醒后如果有第二个参数，如果false，解锁，再次休眠，等待再次被notify_one唤醒
  - 如果为true，返回，此时互斥量被锁
  - 如果没有第二个参数，wait返回
- 1
  2
  3
  4
  void wait(unique_lock<mutex>& _Lck)
  { // wait for signal
  _Cnd_waitX(_Mycnd(), _Lck.mutex()->_Mymtx());
  }
```
template<class _Predicate>
    void wait(unique_lock<mutex>& _Lck, _Predicate _Pred)
    {    // wait for signal and test predicate
    while (!_Pred())
        wait(_Lck);
    }
```
  1
- 1


void outMsgRecvQueue2()
    {
        int cmd=0;
        while(true)
        {
            unique_lock<mutex> ulock(mymutex1);
            mycond.wait(ulock,[this]{
                if(!msgRecvQueue.empty())
                    return true;
                else 
                    return false;
            });
              cmd=msgRecvQueue.front();
            msgRecvQueue.pop_front();
            ulock.unlock();//提前unlock
            cout<<"recvive a msg : "<<cmd<<endl;
            ///wait//
        }
    }

    void inMsgRecvQueue()
    {
        for(int i=0;i<100;i++)
        {
            unique_lock<mutex> ulock(mymutex1);
            msgRecvQueue.push_back(i);
            cout<<"send a msg: "<<i<<endl;
            mycond.notify_one();//尝试wait 线程唤醒
        }
    }

wait 唤醒后获得锁，和inMsgRecvQueue中获得锁，抢占资源，有可能send 多个消息，只有一个通知
唤醒后，如果不是卡在wait等待状态，丢失消息
不一定in 执行一次，Out执行一次
in限流，out多个线程取出数据
如果in先执行，out后执行，in先获得锁，Out不能获得锁，不会进行到wait

3 notify_all()

notify_one只能通知一个线程，通知哪个不确定
醒着获得锁失败于是阻塞
notify_all 唤醒所有正在wait的线程

9 async future packaged_task promise

1 asyc future 创建后台任务并返回值

希望线程返回一个结果
asyc 函数模板，启动一个异步任务，返回future对象，一个类模板
异步任务—自动创建一个线程并开始自动执行对应的线程入口函数，返回的future对象中包含了线程入口函数返回的结果，即线程返回的结果，调用future对象成员函数get返回
future 提供一种访问异步操作结果的机制
- 当get时，会卡在get代码处，get等待线程结束返回结果，get只能获得一次，
- result.wait()等待线程返回，不返回结果
- 当没有get时，卡在主线程的return处，在主线程最后运行子线程
向async传递一个参数，参数类型std::launch类型
- std::launch::defered 表示线程入口函数调用被延迟到std::future的wait或者get()函数调用时才执行
- 如果wait get没有被调用，则不会执行该线程（），实际上，根本没有创建线程
- std::launch::defered 实际上并没有创建线程，在主线程中执行线程入口函数
- std::launch::async 在调用async函数的时候开始创建线程
- std::launch::async|std::launch::defered 默认参数，或者std::launch::async 或者std::launch::defered

1
2
3

X x;
async(&X::foo,&x,1,"test");//1 次构造
async(&X::foo,x,2,"test2");//3 次复制构造

static int mythread()
{
    cout<<"child thread id "<<this_thread::get_id()<<endl;
    chrono::microseconds dura(5000);

    this_thread::sleep_for(dura);

    return 5;
}
void asyc_future_packaged_task_test()
{
    cout<<"asyc_future_packaged_task_test"<<endl;
    cout<<"main thread id :"<<this_thread::get_id()<<endl;
    future<int> result=async(mythread);
    cout<<"continue...."<<endl;//不会卡在这里

    cout<<"result= "<<result.get()<<endl;//当取get时，会卡在这里
	//result.get()// 只能调用一次
}

1
2
3

A a;
future<int >result2=async(&A::mythread,&a,temp);
//&a引用,确保 传递的是同一个a

2 packaged_task

打包任务，把任务打包起来，类模板，模板参数为可调用对象，把各种可调用对象包装起来，方便作为线程入口函数
包装可调用对象为一个类模板对象，传递给thread 时引用ref
packaged_task 包装起来的可调用对象，还可以直接调用，pacaged_task也是一个可调用对象
以便通过get_future获得线程返回值
1

   int mythread2(int num)
{
    cout<<"child thread id "<<this_thread::get_id()<<endl;
    chrono::milliseconds dura(5000);
this_thread::sleep_for(dura);
cout<<"child thread end  num: "<<num<<"id "<<this_thread::get_id()<<endl;
return num;
   }

std::packaged_task<int(int)>mypt(mythread2);//把函数mythread 包装
thread t1(ref(mypt),1);//1 作为mythread的参数
t1.join();
future<int> result4=mypt.get_future();
cout<<"result4= "<<result4.get()<<endl; //立即拿到值

// 借助于lambda 实现packaged_task
 packaged_task<int(int)> mypt2([](int mypar)
    {
   cout<<"child thread id "<<this_thread::get_id()<<endl;                    		chrono::milliseconds dura(5000);

   this_thread::sleep_for(dura);
    cout<<"child thread end  num: "<<mypar<<"id "<<this_thread::get_id()<<endl;
    return mypar;
});

thread t2(ref(mypt2),2);//1 作为mythread的参数
t2.join();
future<int> result5=mypt2.get_future();
cout<<"result4= "<<result5.get()<<endl; //立即拿到值


mypt55(105);//直接调用
 future<int> result6=mypt55.get_future();
cout<<"result4= "<<result6.get()<<endl; //立即拿到值



 cout<<"packaged_task in vector "<<endl;
std::packaged_task<int(int)>mypt4(mythread2);
vector <packaged_task<int(int)>> mytasks;
mytasks.push_back(move(mypt4)); //移动语义
auto iter=mytasks.begin();
std::packaged_task<int(int)>mypt5=move(*iter);
mytasks.erase(iter);//删除之后，iter之后失效，后续代码不可以再用iter,iter为野指针

mypt5(4);
result6=mypt5.get_future();
cout<<"result7= "<<result6.get()<<endl; //立即拿到值

3 promise
在某个线程中给它赋值，在其他线程中取出来
通过promise 保存一个值，在将来某个时刻通过把一个future绑定到promise中来得到这个绑定的值
没有join时，卡在get处，但是还是会有异常，所以thread –join
promise 和future 匹配可实现在两个线程间传递数据
当“承诺”的值已经设置完毕(使用set_value()成员函数)，对应“期望”的状态变为“就绪”，并且可用于检索已存储的值。当你在设置值之前销毁 std::promise ，将会存储一个异常

void promise_thread(promise<int>& tmp,int calc)
{
    calc++;
    calc*=10;
    chrono::milliseconds dura(5000);
    this_thread::sleep_for(dura);
    int result=calc;
    tmp.set_value(result);// 结果保存在promise tmp中
}

 promise<int> myprom;
 thread t11(promise_thread,ref(myprom),123);
  t11.join();
future<int> fut=myprom.get_future();// promise 和future绑定 获取线程返回值
auto result_pro=fut.get();
cout<<"result_pro: "<<result_pro<<endl;
///
 void promise_thread2(future<int>& tmp)
 {
    int result=tmp.get();
    cout<<"promise_thread2 get value "<<result<<endl;
 }

thread t12(promise_thread2,ref(fut));
t12.join();

promise 存储异常
会在future.get时抛出

void promise_thread_square(promise<double>& pr,int x)
{
    try {
        if(x<0)
            throw std::out_of_range("x<0");
  
        cout<<"promise_thread_square  try...."<<endl;
        pr.set_value(sqrt(x));
  
    } catch (...) {
        cout<<"promise_thread_square  catch...."<<endl;
        pr.set_exception(std::current_exception());
        //pr.set_exception(make_exception_ptr(std::logic_error("sqrt x error")));
    }
}

  promise<double> myprom2;
  future<double> fut2=myprom2.get_future();
  thread t21(promise_thread_square,ref(myprom2),-1);
  t21.join();
  cout<<"promise thread over!"<<endl;
  double res=fut2.get();
  cout<<"promise store exception over res= "<<res<<endl;

4 小结

通过future.get() 获得线程返回值，
- 主线程中 asyc函数模板直接返回future 对象
  - 第一个参数 launch::defered ,线程延迟到future::wait /get执行，否则不执行，且在主线程中执行
  - 默认第一个参数launch::async，线程立即执行
- 主线程中 packaged_task 类模板，packaged_task::get_future获得future 对象
  - 可直接调用
  - thread时， thread t2(ref(mypt2),2); //1 作为mythread的参数
  - future result5=mypt2.get_future(); thread t2(move(mypt2),2); move之前获得future
- 主/另一个子线程中 promise 类模板 promise.set_value保存值，myprom.get_future()在另一个线程中获得值
  - 可用于两个线程间交互数据

10 future其他成员函数 shared_future atomic

0 时钟

时钟当前时间可以通过调用静态成员函数now()从时钟类中获取
- std::chrono::system_clock 系统时钟
- std::chrono::steady_clock 稳定时钟
- std::chrono::high_resolution_clock 标准库中提供的具有最小节拍周期(因此具有最高的精度[分辨率])的时钟

时延

连续时间用duration<long long, ratio<1,1000> > //第一个模板参数是一个类型
表示(比如，int，long或double) 第二个模板参数ratio<1,1000> 每一个单元所用秒数

typedef ratio<1, 1000> milli;

typedef duration<long long, nano> nanoseconds;
typedef duration<long long, micro> microseconds;
typedef duration<long long, milli> milliseconds;
typedef duration<long long> seconds;
typedef duration<int, ratio<60> > minutes;
typedef duration<int, ratio<3600> > hours;

std::chrono::milliseconds ms(54802);
std::chrono::seconds s=
std::chrono::duration_cast<std::chrono::seconds>(ms);//截断为 54s


std::chrono::milliseconds(1234).count()//时延中可以通过count()成员函数获得单位时间的数量 1234

时间点
- 时间点可以用 std::chrono::time_point<> 的类型模板实例来表示，实例的第一个参数
  用来指定所要使用的时钟，第二个函数参数用来表示时间的计量单位(特化
  的 std::chrono::duration<> )
wait_for wait_until

std::this_thread namespace	sleep_for(duration) sleep_until (time_point)	N/A
std::condition_ variable or std::condition_ variable_any	wait_for(lock, duration) wait_until(lock, time_point)	std::cv_status:: timeout or std::cv_status:: no_timeout
	wait_for(lock, duration, predicate) wait_until(lock, time_point, predicate)	bool—唤醒时，返回谓词结果
std::timed_mutex or std::recursive_ timed_mutex	try_lock_for (duration) try_lock_until (time_point)	bool—获取锁，返回true，否则false
std::unique_ lock	unique_lock(lockable, duration) unique_lock(lockable, time_point)	N/A—对新构建的对像调用owns_lock() returns true if the lock was acquired, false otherwise
	try_lock_for(duration) try_lock_until (time_point)	bool—获取锁，返回true，否则false
std::future or std::shared_ future	wait_for(duration) wait_until (time_point)	std::future_status:: timeout if the wait timed out, std::future_ status::ready if the future is ready, or std::future_status:: deferred if the future holds a deferred function that hasn’t yet started

1 future其他成员函数

future_status 枚举状态


enum class future_status {	// names for timed wait function returns
  	ready,
  	timeout,
  	deferred
  	};

timeout 线程执行时间超时，线程还没执行完，
ready 线程成功返回
defered 延迟，如果async第一个参数设置为std::launch::async满足条件

int mythread()
{
    cout<<"mythread start id "<<this_thread::get_id()<<endl;
    chrono::milliseconds dura(2500);
    this_thread::sleep_for(dura);
    cout<<"mythread stop id "<<this_thread::get_id()<<endl;
    return 7;
}

    void future_member_func_atomic_test()
    {
        cout<<"future_member_func_atomic_test id "<<this_thread::get_id()<<endl;
       // future<int> result=async(mythread);
        future<int> result=async(launch::deferred,mythread);
       // cout<<"result: "<<result.get()<<endl;

        future_status status=result.wait_for(std::chrono::milliseconds(3500));//1000 ms时，timeout
        if(status==future_status::timeout)
        {
            cout<<"status timeout! thread not over"<<endl;
        }
        else if(status==future_status::ready)
        {
            cout<<"status ready! thread  over "<<result.get()<<endl;
        }
        else if(status==future_status::deferred)
        {
            cout<<"status deferred! thread  delay to execute "<<endl;
            cout<<"result.get(): "<<result.get()<<endl;
        }


        return ;


    }

2 shared_future

int mythread2(int num)
{
    cout<<"mythread2 start id "<<this_thread::get_id()<<endl;
    chrono::milliseconds dura(2500);
    this_thread::sleep_for(dura);
    cout<<"mythread2 stop id "<<this_thread::get_id()<<endl;
    return num;
}
void mythread22(future<int>& ft)
{
     cout<<"mythread22 start id "<<this_thread::get_id()<<endl;
    int cmd=ft.get();
     cout<<"mythread22 stop  cmd "<<cmd<<" id "<<this_thread::get_id()<<endl;

}
  packaged_task<int(int)> mypt(mythread2);
    future<int> result2=mypt.get_future();

    thread t1(ref(mypt),12);
    t1.join();

    thread t2(mythread22,ref(result2));
    t2.join();

future get两次会异常，get函数的设计是一个移动语义
多个线程获得值时,遇到问题
shared_future 类模板，get 函数设计复制数据
result.valid() ，检测当前future对象，能否get到数据

int mythread2(int num)
{
    cout<<"mythread2 start id "<<this_thread::get_id()<<endl;
    chrono::milliseconds dura(2500);
    this_thread::sleep_for(dura);
    cout<<"mythread2 stop id "<<this_thread::get_id()<<endl;
    return num;
}

void mythread222(shared_future<int>& ft)
{
     cout<<"mythread222 shared_future start id "<<this_thread::get_id()<<endl;
    int cmd=ft.get();
     cout<<"mythread222 shared_future stop  cmd "<<cmd<<" id "<<this_thread::get_id()<<endl;

}

packaged_task<int(int)> mypt2(mythread2);
future<int> result_temp=mypt2.get_future();

shared_future<int> shared_result(move(result_temp));//右值
//shared_future<int> shared_result2(result_temp.share());//右值
//shared_future<int> shared_result3(mypt2.get_future());// 通过future 构造shared_future
if(!result_temp.valid())
{
    cout<<"result_temp  invalid"<<endl;
}
thread t3(ref(mypt2),13);
t3.join();
// cout<<"shared_result: "<<shared_result.get()<<endl;

thread t4(mythread222,ref(shared_result));
t4.join();

3 atomic原子操作

互斥量多线程编程中，保护共享数据
两个线程，对一个变量操作，一个写，一个读

int atomvalue=5;
//read
int tmpvalue=atomvalue;

//write
atomvalue=6;

有可能读到的是 5 6 “之间” 的值
加法没有被执行完毕，就被其他线程打断，
可通过加锁解决，但是效率慢

int g_mycout=0;
mutex g_mymutex;
void atomic_write()
{
    for(int i=0;i<1000000000;i++)
    {
        //g_mymutex.lock();
        g_mycout++;
        //g_mymutex.unlock();
	}
    return;
}
  thread mytob1(atomic_write);
    thread mytob2(atomic_write);
    mytob1.join();
    mytob2.join();

    cout<<"thread 1 2 over g_mycout: "<<g_mycout<<endl;

原子操作，无锁技术不用互斥量，的多线程并发，在多线程中不会被打断的程序执行片段
原子操作效率更高
互斥量加锁一般针对一个代码段（几行代码），原子操作针对的是一个变量
原子操作–不可分割的操作，要么完成，要么没有完成，不可能中间状态
std::atomic 一个类模板，用来封装某个类型的值
atomic 赋值构造函数删除了！！！！！

std::atomic<int> g_mycout2(0);
atomic<int> g_mycout2=0;/////error!!!!
void atomic_write2()
{
    for(int i=0;i<10000000;i++)
    {
        g_mycout2++;
    }
    return;
}
thread mytob11(atomic_write2);
thread mytob21(atomic_write2);
mytob11.join();
mytob21.join();


void atomic_bool_read()
{
    chrono::milliseconds dura(1000);
    while(g_ifend==false)
    {
        cout<<"id "<<this_thread::get_id()<<" is running !"<<endl;
        this_thread::sleep_for(dura);
    }
    cout<<"id "<<this_thread::get_id()<<" has exited! "<<endl;
    return;
}

thread mytob00(atomic_bool_read);
    thread mytob01(atomic_bool_read);
    

    this_thread::sleep_for(chrono::milliseconds(5000));
    g_ifend=true;
    mytob00.join();
    mytob01.join();
    cout<<"atomic write end!"<<endl;

原子操作一般用于统计，计数
atomic作为成员变量时，std::atomic g_mycout2(0)；error？？？？
作为全局变量时，std::atomic g_mycout2(0) ok

11atomic-again，async-again

1 atomic-again

atomic<int> g_mycount(0);
  void atomic_int_write()
  {
 for(int i=0;i<10000000;i++)
 {
     g_mycount+=1;
     g_mycount++;
     g_mycount=g_mycount+1;// error
 }
  
 return;
  }

thread t1(atomic_int_write);
thread t2(atomic_int_write);
t1.join();
 t2.join();
 cout<<"g_mycount: "<<g_mycount<<endl;

一般atomic原子操作针对++ – += -= &= |=支持，单目运算符。其他可能不支持

2 async-again

async参数
- launch::defered 延迟调用线程函数到future::wait future::get，且不创建子线程，直接在主线程调用，如果没有调用future::wait future::get，线程入口函数不执行
- launch::async 默认参数，强制创建线程，线程入口函数在新线程中运行
- 系统资源紧张时，创建线程失败，执行thread()时，整个程序可能崩溃
- async（）一般不叫创建线程，（尽管也可以创建了线程），叫创建异步任务
- 同时 launch::async() | launch::defered 时，创建新线程，async可能 launch::async() 或者 launch::defered，是否创建线程不确定，系统自行决定异步或同步方式运行
- 默认参数 launch::async() | launch::defered 。
async和thread区别
- thread 创建线程，如果系统资源紧张，创建线程失败，崩溃
- thread创建线程的方式，拿到线程返回值，需要全局变量或。。。。
- async 创建异步任务，可创建也可不创建线程，可通过直接返回future对象拿到线程返回值
- 系统资源紧张，async 默认参数就会不创建线程以同步方式在调用get,wait函数的线程运行线程函数
- 可通过参数launch::async强制创建线程

async不确定问题的解决

future result=async(mythread) ,整个异步任务是否有没被推迟执行
async有没有被创建新线程

通过futrue_status status=result.wait_for(0s); 通过future等待0s获得future_status


future<int >result=async(mythread);
       future_status status=result.wait_for(0s);
       if(status==future_status::deferred)//未创建线程
       {
           cout<<"deferred"<<endl;
           cout<<"result.get(): "<<result.get()<<endl;
       }
       else if(status==future_status::ready)//线程已经执行完毕
       {
           cout<<"ready"<<endl;
           cout<<"result.get(): "<<result.get()<<endl;
       }
       else if(status==future_status::timeout)//线程还未执行完毕
       {
           cout<<"timeout"<<endl;
           cout<<"result.get(): "<<result.get()<<endl;
       }

future_status::deferred	The function to calculate the result has not been started yet	线程函数延迟执行
future_status::ready	The result is ready	线程已经执行完毕结果
future_status::timeout	The timeout has expired	线程仍在执行

12 windows临界区，其他各种Mutex互斥量

1 windows 临界区

#include <Windows.h>
#define __WINDOWS__

namespace other_mutex {

class A{
private:
    list<int > msgRecvQueue;
    mutex mymutex;
#ifdef  __WINDOWS__
    CRITICAL_SECTION mywinsec;// windows临界区，用之前必须初始化
#endif

public:
    A()
    {

#ifdef __WINDOWS__
    InitializeCriticalSection(&mywinsec);
#endif
    }

    bool msgRecvProc(int& cmd)
    {
        bool res;
#ifdef __WINDOWS__
        EnterCriticalSection(&mywinsec);//进入临界区 --加锁
#else
        mymutex.lock();
#endif
        if(msgRecvQueue.empty())
        {
            res= false;
        }
        else
        {
            cmd= msgRecvQueue.front();
            msgRecvQueue.pop_front();
            res= true;
        }
#ifdef __WINDOWS__
        LeaveCriticalSection(&mywinsec);//离开临界区  解锁
#else
         mymutex.unlock();
#endif
        return res;

    }
    void outMsgRecvQueue()
    {
        int cmd;
        for(int i=0;i<1000;i++)
        {
            if(msgRecvProc(cmd))
            {
                cout<<"rec a msg cmd: "<<cmd<<endl;
            }
            else
            {
                cout<<"list empty"<<endl;
            }
        }
    }
    void inMsgRecvQueue()
    {
        for(int i=0;i<1000;i++)
        {

#ifdef __WINDOWS__
            EnterCriticalSection(&mywinsec);//进入临界区 加锁
            EnterCriticalSection(&mywinsec);//可以多次进入
#else
           mymutex.lock();
#endif
            msgRecvQueue.push_back(i);

#ifdef __WINDOWS__
            LeaveCriticalSection(&mywinsec);//离开临界区 解锁
            LeaveCriticalSection(&mywinsec);//多次进入，多次离开
#else
            mymutex.unlock();
#endif
            cout<<"send a msg "<<i<<endl;
        }
    }
};
    void other_mutex_test()
    {
        cout<<"other_mutex test id "<<this_thread::get_id()<<endl;

        A a;
        thread read_thread(&A::outMsgRecvQueue,&a);
        thread write_thread(&A::inMsgRecvQueue,&a);

        read_thread.join();
        write_thread.join();
        return ;
    }

2 多次进入临界区试验

同一个线程中，多次进去同一个临界区变量代表的临界区可以，不会等待
但是，几次进去，就要几次离开
而C++11不允许同一个线程中同一个mutex多次加锁

3 自动析构技术

lock_guard sbguard; unque_lock sbfuard;
RAII 类 (resource acquisition is initialization) 资源获取即初始化
类似的有智能指针，容器

4 recursive_mutex递归的独占互斥量

//可能存在情况
void test1()
{
   lock_guard<mutex> sbguard(mymutex); 
}
void test2()
{
     lock_guard<mutex> sbguard(mymutex); 
     test1(); //锁了两次
}

mutex 独占互斥量，只有一个线程可以拿到锁
recursive_mutex 递归的独占互斥量，允许同一个线程，同一个互斥量，多次lock
和mutex 用法相同，recursive_mutex 效率低

1	lock_guard<recursive_mutex> lg(mymutex);

5 带超时的互斥量 std::timed_mutex recursive_timed_mutex

std::timed_mutex 超时的独占互斥量，不会一直阻塞

try_lock_for 参数为时间，等待一段时间，如果拿到锁，或者等待超时时间没拿到锁，就返回

chrono::milliseconds timeout(100);
           if(mymutex.try_lock_for(timeout))
           {//100ms之内拿到锁
               msgRecvQueue.push_back(i);
               mymutex.unlock();
           }
           else //100ms 没有拿到锁
           {
               cout<<" has not obtain lock!!"<<endl;
               chrono::milliseconds sleep(100);
               this_thread::sleep_for(sleep);
           }

try_lock_until 参数为未来的一个时间点，未来的时间没到的时间段内，如果拿到锁，流程就走下来，如果时间到，没有拿到锁，流程也走下来

chrono::milliseconds timeout(1);
          //if(mymutex.try_lock_for(timeout))
          if(mymutex.try_lock_until(chrono::steady_clock::now()+timeout))
          {//100ms之内拿到锁
              msgRecvQueue.push_back(i);
               cout<<"send a msg "<<i<<endl;
             mymutex.unlock();
  
          }
          else //100ms 没有拿到锁
          {
              cout<<" has not obtain lock!!"<<endl;
              chrono::milliseconds sleep(100);
              this_thread::sleep_for(sleep);
          }

recursive_timed_mutex 带超时功能的递归独占互斥量（允许同一个线程多次获取同一个互斥量）

13 补充，线程池、线程数量、总结

1 补充

condition_variable::wait
- 如果pred为真，直接返回
- 如果pred为假，解锁互斥量，并阻塞到本行，一直到其他某个线程norify_One()成员函数
- 没有Pred，和Pred为假相同
- 如果唤醒，
  - 尝试获得锁，失败则阻塞
  - 如果获得锁成功，但是Pred 为假，则解锁，等待被再次唤醒
  - 没有第二个参数，直接返回
- wait执行完毕，一定是加锁+被唤醒
虚假唤醒
- 存在队列中没有数据，被唤醒 —–唤醒但是Pred为假
- 当多个notify_one迭代一起时
- 所以要用第二个参数Pred，再次确认，唤醒后，能否执行一系列动作
- wait中第二个参数Pred，判断公共数据是否存在

atomic

1
2
3

//读atm是个原子操作，但是整个一行代码不是原子操作
//atm的值可能是之前的值
cout<<atm<<endl;//

原子操作不能赋值

//以原子方式读atm值
atmoic<int> atm2(atm.load());
//原子操作写值
atm2.store(12);

2 线程池

场景
- 每来一个客户端，就创建一个线程为该客户端提供服务
- 客户端过多时？？？？？？
- 程序偶尔创建一个线程，稳定性不高
- 统一调度一堆线程，统一管理，循环利用
实现方式
- 程序启动时，一次性创建好一定数量的线程

3 线程数量

数量极限问题，2000个？？？
开放商的建议
创建多个线程完成业务，一个线程一条通路，100个要充值，创建110个线程

4 总结

C++多线程真的变态。。

本文标题:C++11 多线程编程

文章作者:bliss7

发布时间:2019-01-03, 00:00:00

最后更新:2019-07-23, 13:36:04

原始链接:blissseven.top/2019/01/03/concurrency-in-cpp11/

许可协议: "署名-非商用-相同方式共享 4.0" 转载请保留原文链接及作者。