Glib学习（15）线程之间的异步通信 Asynchronous Queues

偏执的太偏执、 2022-05-31 07:39 197阅读 0赞

glib源码下载：http://ftp.gnome.org/pub/gnome/sources/glib/

glib帮助文档：https://developer.gnome.org/glib/

本节主要讲述线程间异步通信，实现原理就是用线程锁和队列实现的线程异步队列。

# 描述 #

通常你需要在不同的线程之间进行通信。 一般来说，不要通过共享内存来做这件事，而应该通过明确的消息传递。 这些消息对于多线程应用程序只是异步的，因为同步操作也可以在同一个线程中完成。  
异步队列是大多数其他GLib数据结构的一个例外，因为它们可以在多个线程中同时使用而不需要显式锁定，并且它们自带内置的引用计数。 这是因为异步队列的性质是它总是会被至少两个并发线程使用。  
对于使用异步队列，您首先必须使用g\_async\_queue\_new（）创建一个。 GAsyncQueue结构是引用计数，使用g\_async\_queue\_ref（）和g\_async\_queue\_unref（）来管理您的引用。  
想要发送消息到该队列的线程只需调用g\_async\_queue\_push（）将消息推送到队列。  
预期来自异步队列消息的线程只需调用该队列的g\_async\_queue\_pop（）。 如果队列中没有可用的消息，那么线程现在处于休眠状态，直到消息到达。 该消息将从队列中移除并返回。 函数g\_async\_queue\_try\_pop（）和g\_async\_queue\_timeout\_pop（）可以用来检查消息的存在或者仅仅等待消息的一定时间。  
几乎每个函数都有两个变量，一个锁定队列，另一个不锁定队列。 这样，您可以保持队列锁定（通过g\_async\_queue\_lock（）获取并通过g\_async\_queue\_unlock（））通过多个队列访问指令进行释放。 这对确保队列的完整性可能是必要的，但只有在真正需要时才能使用，因为如果使用不当，可能会使你的生活变得更加困难。 通常你只能使用锁定函数变体（那些没有\_unlocked后缀的变体）。  
在许多情况下，当您需要将工作分配到一组工作线程而不是手动使用GAsyncQueue时，使用GThreadPool可能会更方便。 GThreadPool在内部使用GAsyncQueue。

这个队列主要分为两套，一套是不使用线程队列锁时使用的函数，一套是使用线程锁时使用的函数，这两套不能混用，因为混用可能造成死锁。

不带锁相关函数

GAsyncQueue *	g_async_queue_ref ()
    void	g_async_queue_unref ()
    void	g_async_queue_push ()
    void	g_async_queue_push_sorted ()
    void	g_async_queue_push_front ()
    gboolean	g_async_queue_remove ()
    gpointer	g_async_queue_pop ()
    gpointer	g_async_queue_try_pop ()
    gpointer	g_async_queue_timeout_pop ()
    gint	g_async_queue_length ()
    void	g_async_queue_sort ()

带锁相关函数

void	g_async_queue_lock ()
    void	g_async_queue_unlock ()
    void	g_async_queue_ref_unlocked ()
    void	g_async_queue_unref_and_unlock ()
    void	g_async_queue_push_unlocked ()
    void	g_async_queue_push_sorted_unlocked ()
    void	g_async_queue_push_front_unlocked ()
    gboolean	g_async_queue_remove_unlocked ()
    gpointer	g_async_queue_pop_unlocked ()
    gpointer	g_async_queue_try_pop_unlocked ()
    gpointer	g_async_queue_timeout_pop_unlocked ()
    gint	g_async_queue_length_unlocked ()
    void	g_async_queue_sort_unlocked ()

下面是函数翻译

GAsyncQueue \*g\_async\_queue\_new ()  
创建一个新的异步队列。  
  
返回  
一个新的GAsyncQueue。 使用g\_async\_queue\_unref（）释放  
  
GAsyncQueue \*g\_async\_queue\_new\_full ()  
创建一个新的异步队列并设置一个destroy通知函数，该函数用于在队列在最终unref之后销毁时释放任何剩余的队列项。  
  
参数  
item\_free\_func  
函数释放队列元素  
   
返回  
一个新的GAsyncQueue。 使用g\_async\_queue\_unref（）释放  
从：2.16  
  
GAsyncQueue \*g\_async\_queue\_ref ()  
将异步队列的引用计数增加1。您不需要保持锁来调用此函数。  
  
参数  
queue  
一个GAsyncQueue  
   
返回  
传入的队列（从2.6开始）  
  
voidg\_async\_queue\_unref ()  
将异步队列的引用计数减1  
如果引用计数为0，则队列将被销毁并分配的内存将被释放。 所以你不能使用队列，因为它可能已经消失了。 你不需要保持锁来调用这个函数。  
  
参数  
queue  
一个GAsyncQueue。  
  
voidg\_async\_queue\_push ()  
将数据推入队列。 数据不能为NULL。  
  
参数  
queue  
一个GAsyncQueue  
data  
数据推入队列  
  
voidg\_async\_queue\_push\_sorted ()  
使用func将数据插入队列以确定新的位置。  
这个函数需要在推入新元素之前对队列进行排序，请参阅g\_async\_queue\_sort（）。  
这个函数会在对队列进行排序之前锁定队列，并在完成时将其解锁。  
有关func的示例，请参阅g\_async\_queue\_sort（）。  
  
参数  
queue  
一个GAsyncQueue  
data  
将数据推入队列  
func  
GCompareDataFunc用于对队列进行排序  
user\_data  
用户数据传递给func。  
从：2.10  
  
voidg\_async\_queue\_push\_front ()  
将项目推入队列。 项目不能为NULL。 与g\_async\_queue\_push（）不同的是，这个函数在队列中已经存在的项目之前压入新的项目，这样它将成为下一个从队列中弹出的项目。  
  
参数  
queue  
一个GAsyncQueue  
item  
数据推入队列  
由于：2.46  
  
gbooleang\_async\_queue\_remove ()

从队列中删除一个项目。

参数  
queue  
一个GAsyncQueue  
item  
要从队列中删除的数据  
   
返回  
如果项目被移除，则为TRUE  
从：2.46  
  
gpointer g\_async\_queue\_pop ()  
从队列中弹出数据。 如果队列为空，则此功能会阻塞，直到数据可用。  
  
参数  
queue  
一个GAsyncQueue  
  
返回  
来自队列的数据  
  
gpointer g\_async\_queue\_try\_pop ()  
试图从队列中弹出数据。 如果没有数据可用，则返回NULL。  
  
参数  
queue  
一个GAsyncQueue  
   
返回  
来自队列的数据或NULL，当没有数据立即可用时。  
  
gpointer g\_async\_queue\_timeout\_pop ()  
从队列中弹出数据。 如果队列为空，则超时微秒或直到数据可用。  
如果在超时之前没有收到数据，则返回NULL。  
  
参数  
queue  
一个GAsyncQueue  
timeout  
等待的微秒数  
   
返回  
在超时之前没有收到数据时，队列中的数据或NULL。  
  
gint g\_async\_queue\_length ()  
返回队列的长度。  
实际上这个函数返回队列中数据项的数量减去等待线程的数量，所以负值表示等待线程，正值表示队列中可用的项。 返回值0可以表示队列中的n个条目和n个等待的线程。 这可能是由于队列锁定或调度造成的。  
  
参数  
queue  
一个GAsyncQueue。  
   
返回  
队列的长度  
  
void g\_async\_queue\_sort ()  
使用func对队列进行排序。  
sort函数func传递队列的两个元素。 如果它们相等，它应该返回0，如果第一个元素应该在队列中更高，则返回负值;如果队列中第一个元素应该比第二个元素低，则返回正值。  
这个函数会在对队列进行排序之前锁定队列，并在完成时将其解锁。  
如果您正在排序优先级编号列表以确保最低优先级位于队列顶部，则可以使用：  
gint32 id1;  
gint32 id2;  
  
id1 = GPOINTER\_TO\_INT (element1);  
id2 = GPOINTER\_TO\_INT (element2);  
  
return (id1 > id2 ? +1 : id1 == id2 ? 0 : -1);  
参数  
queue  
一个GAsyncQueue  
func  
GCompareDataFunc用于对队列进行排序  
user\_data  
用户数据传递给func  
从：2.10  
  
void g\_async\_queue\_lock ()  
获取队列的锁。 如果另一个线程已经持有该锁，则该呼叫将被阻塞，直到该锁可用。  
调用g\_async\_queue\_unlock（）再次放下锁。  
在按住锁的同时，您只能调用队列上的g\_async\_queue \_ \* \_ unlocked（）函数。 否则，可能会发生死锁。  
  
参数  
queue  
一个GAsyncQueue  
  
void g\_async\_queue\_unlock ()  
释放队列的锁定。  
当你没有获得g\_async\_queue\_lock（）时调用这个函数会导致未定义的行为。  
  
参数  
queue  
一个GAsyncQueue  
  
void g\_async\_queue\_ref\_unlocked ()  
从版本2.8开始，g\_async\_queue\_ref\_unlocked已被弃用，不应在新编写的代码中使用。  
引用计数是自动完成的。 所以不管队列的锁，都可以使用g\_async\_queue\_ref（）。  
将异步队列的引用计数增加1。  
  
参数  
queue  
一个GAsyncQueue  
  
void g\_async\_queue\_unref\_and\_unlock ()  
g\_async\_queue\_unref\_and\_unlock从版本2.8开始已被弃用，不应该用在新编写的代码中。  
引用计数是自动完成的。 所以不管队列的锁，都可以使用g\_async\_queue\_unref（）。  
将异步队列的引用计数减1，并释放锁定。 这个函数必须在持有队列锁的时候被调用。 如果引用计数为0，则队列将被销毁并分配的内存将被释放。  
  
参数  
queue  
一个GAsyncQueue  
  
void g\_async\_queue\_push\_unlocked ()  
将数据推入队列。 数据不能为NULL。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
data  
数据推入队列  
  
void g\_async\_queue\_push\_sorted\_unlocked ()  
使用func将数据插入队列以确定新的位置。  
sort函数func传递队列的两个元素。 如果它们相等，它应该返回0，如果第一个元素应该在队列中更高，则返回负值;如果队列中第一个元素应该比第二个元素低，则返回正值。  
这个函数需要在推入新元素之前对队列进行排序，请参阅g\_async\_queue\_sort（）。  
这个函数必须在持有队列锁的时候被调用。  
有关func的示例，请参阅g\_async\_queue\_sort（）。  
  
参数  
queue  
一个GAsyncQueue  
data  
将数据推入队列  
func  
GCompareDataFunc用于对队列进行排序  
user\_data  
用户数据传递给func。  
从：2.10  
  
void g\_async\_queue\_push\_front\_unlocked ()  
将项目推入队列。 项目不能为NULL。 与g\_async\_queue\_push\_unlocked（）相比，此函数将新项目推到队列中已有的项目之前，这样它将成为下一个从队列中弹出的项目。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
item  
数据推入队列  
由于：2.46  
  
gboolean g\_async\_queue\_remove\_unlocked ()  
从队列中删除一个项目。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
item  
要从队列中删除的数据  
   
返回  
如果项目被移除，则为TRUE  
由于：2.46  
  
gpointer g\_async\_queue\_pop\_unlocked ()  
从队列中弹出数据。 如果队列为空，则此功能会阻塞，直到数据可用。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
   
返回  
来自队列的数据。  
  
gpointer g\_async\_queue\_try\_pop\_unlocked ()  
试图从队列中弹出数据。 如果没有数据可用，则返回NULL。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
   
返回  
来自队列的数据或NULL，当没有数据立即可用时。  
  
gpointer g\_async\_queue\_timeout\_pop\_unlocked ()  
从队列中弹出数据。 如果队列为空，则超时微秒或直到数据可用。  
如果在超时之前没有收到数据，则返回NULL。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
timeout  
等待的微秒数  
   
返回  
在超时之前没有收到数据时，队列中的数据或NULL。  
  
gint g\_async\_queue\_length\_unlocked ()  
返回队列的长度。  
实际上这个函数返回队列中数据项的数量减去等待线程的数量，所以负值表示等待线程，正值表示队列中可用的项。 返回值0可以表示队列中的n个条目和n个等待的线程。 这可能是由于队列锁定或调度造成的。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
   
返回  
队列的长度。  
  
void g\_async\_queue\_sort\_unlocked ()  
使用func对队列进行排序。  
sort函数func传递队列的两个元素。 如果它们相等，它应该返回0，如果第一个元素应该在队列中更高，则返回负值;如果队列中第一个元素应该比第二个元素低，则返回正值。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
func  
GCompareDataFunc用于对队列进行排序  
user\_data  
用户数据传递给func  
从：2.10  
  
gpointer g\_async\_queue\_timed\_pop ()  
g\_async\_queue\_timed\_pop已弃用，不应在新编写的代码中使用。  
使用g\_async\_queue\_timeout\_pop（）。  
从队列中弹出数据。 如果队列为空，则阻塞直到end\_time或直到数据可用。  
如果在end\_time之前没有收到数据，则返回NULL。  
为了便于计算end\_time，可以使用g\_get\_current\_time（）和g\_time\_val\_add（）的组合。  
  
参数  
queue  
一个GAsyncQueue  
end\_time  
一个GTimeVal，确定最后的时间  
   
返回  
在end\_time之前未收到数据时，队列中的数据或NULL。  
  
gpointer g\_async\_queue\_timed\_pop\_unlocked ()  
不推荐使用g\_async\_queue\_timed\_pop\_unlocked，不应该在新编写的代码中使用。  
使用g\_async\_queue\_timeout\_pop\_unlocked（）。  
从队列中弹出数据。 如果队列为空，则阻塞直到end\_time或直到数据可用。  
如果在end\_time之前没有收到数据，则返回NULL。  
为了便于计算end\_time，可以使用g\_get\_current\_time（）和g\_time\_val\_add（）的组合。  
这个函数必须在持有队列锁的时候被调用。  
  
参数  
queue  
一个GAsyncQueue  
end\_time  
一个GTimeVal，确定最后的时间  
   
返回  
在end\_time之前未收到数据时，队列中的数据或NULL。

下面是一个简单的例程：

#include <glib.h>
    #include <glib/gprintf.h>
    
    GAsyncQueue *async_queue = NULL;
    
    gpointer thread_func1 (gpointer data)
    {
    	gint *usr_data;
    	while(1)
    	{
    		g_printf ("%s in\n", __func__);
    		g_async_queue_lock (async_queue);
    		usr_data = (gint *)g_async_queue_pop_unlocked (async_queue);
    		g_printf("%s pop: %d\n", __func__, *usr_data);
    		g_free (usr_data);
    		g_printf ("queue length %d\n", g_async_queue_length_unlocked (async_queue));
    		g_async_queue_unlock (async_queue);
    		g_usleep (2000000);
    	}
    }
    
    gpointer thread_func2 (gpointer data)
    {
    	gint *tmp;
    	gint count = 0;
    	while(1)
    	{
    		g_printf ("%s in\n", __func__);
    		g_async_queue_lock (async_queue);
    		tmp = (gint *)g_new0(gint, 1);
    		*tmp = count++;
    		g_async_queue_push_unlocked (async_queue, tmp);
    		g_printf("%s count: %d\n", __func__, *tmp);
    		g_async_queue_unlock (async_queue);
    		g_usleep (1000000);
    	}
    }
    
    
    int main(int argc, char **argv)
    {
    	g_printf ("%s in\n", __func__);
    
    	GThread *gthread1 = NULL, *gthread2 = NULL;
    	async_queue = g_async_queue_new ();
    	gthread2 = g_thread_new ("func2", thread_func2, NULL);
    	g_usleep (1000000);
    	gthread1 = g_thread_new ("func1", thread_func1, NULL);
    	g_thread_join (gthread1);
    	g_thread_join (gthread2);
    	
    	g_printf ("%s out\n", __func__);
        return 0;
    }