AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

linux下C语言实现多线程通信—环形缓冲区,可用于生产者(producer)/消费者(consumer)【转】

2017-12-21 2592
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介：

转自：http://blog.chinaunix.net/uid-28458801-id-4262445.html

操作系统：ubuntu10.04

前言：
    在嵌入式开发中，只要是带操作系统的，在其上开发产品应用，基本都需要用到多线程。
    为了提高效率，尽可能的提高并发率。因此，线程之间的通信就是问题的核心。
    根据当前产品需要，使用 环形缓冲区 解决。

一，环形缓冲区的实现
    1，cbuf.h

点击(此处)折叠或打开

  1. #ifndef __CBUF_H__
  2. #define __CBUF_H__

  4. #ifdef __cplusplus
  5. extern "C" {
  6. #endif

  8. /* Define to prevent recursive inclusion 
  9. -------------------------------------*/
  10. #include "types.h"
  11. #include "thread.h"


  14. typedef    struct _cbuf
  15. {
  16.     int32_t        size;            /* 当前缓冲区中存放的数据的个数 */
  17.     int32_t        next_in;        /* 缓冲区中下一个保存数据的位置 */
  18.     int32_t        next_out;        /* 从缓冲区中取出下一个数据的位置 */
  19.     int32_t        capacity;        /* 这个缓冲区的可保存的数据的总个数 */
  20.     mutex_t        mutex;            /* Lock the structure */
  21.     cond_t        not_full;        /* Full -> not full condition */
  22.     cond_t        not_empty;        /* Empty -> not empty condition */
  23.     void        *data[CBUF_MAX];/* 缓冲区中保存的数据指针 */
  24. }cbuf_t;


  27. /* 初始化环形缓冲区 */
  28. extern    int32_t        cbuf_init(cbuf_t *c);

  30. /* 销毁环形缓冲区 */
  31. extern    void        cbuf_destroy(cbuf_t    *c);

  33. /* 压入数据 */
  34. extern    int32_t        cbuf_enqueue(cbuf_t *c,void *data);

  36. /* 取出数据 */
  37. extern    void*        cbuf_dequeue(cbuf_t *c);


  40. /* 判断缓冲区是否为满 */
  41. extern    bool        cbuf_full(cbuf_t    *c);

  43. /* 判断缓冲区是否为空 */
  44. extern    bool        cbuf_empty(cbuf_t *c);

  46. /* 获取缓冲区可存放的元素的总个数 */
  47. extern    int32_t        cbuf_capacity(cbuf_t *c);


  50. #ifdef __cplusplus
  51. }
  52. #endif

  54. #endif
  55. /* END OF FILE 
  56. ---------------------------------------------------------------*/


    2，cbuf.c

点击(此处)折叠或打开

  1. #include "cbuf.h"



  5. /* 初始化环形缓冲区 */
  6. int32_t        cbuf_init(cbuf_t *c)
  7. {
  8.     int32_t    ret = OPER_OK;

  10.     if((ret = mutex_init(&c->mutex)) != OPER_OK)    
  11.     {
  12. #ifdef DEBUG_CBUF
  13.     debug("cbuf init fail ! mutex init fail !\n");
  14. #endif
  15.         return ret;
  16.     }

  18.     if((ret = cond_init(&c->not_full)) != OPER_OK)    
  19.     {
  20. #ifdef DEBUG_CBUF
  21.     debug("cbuf init fail ! cond not full init fail !\n");
  22. #endif
  23.         mutex_destroy(&c->mutex);
  24.         return ret;
  25.     }

  27.     if((ret = cond_init(&c->not_empty)) != OPER_OK)
  28.     {
  29. #ifdef DEBUG_CBUF
  30.     debug("cbuf init fail ! cond not empty init fail !\n");
  31. #endif
  32.         cond_destroy(&c->not_full);
  33.         mutex_destroy(&c->mutex);
  34.         return ret;
  35.     }

  37.     c->size     = 0;
  38.     c->next_in    = 0;
  39.     c->next_out = 0;
  40.     c->capacity    = CBUF_MAX;

  42. #ifdef DEBUG_CBUF
  43.     debug("cbuf init success !\n");
  44. #endif

  46.     return ret;
  47. }


  50. /* 销毁环形缓冲区 */
  51. void        cbuf_destroy(cbuf_t    *c)
  52. {
  53.     cond_destroy(&c->not_empty);
  54.     cond_destroy(&c->not_full);
  55.     mutex_destroy(&c->mutex);

  57. #ifdef DEBUG_CBUF
  58.     debug("cbuf destroy success \n");
  59. #endif
  60. }



  64. /* 压入数据 */
  65. int32_t        cbuf_enqueue(cbuf_t *c,void *data)
  66. {
  67.     int32_t    ret = OPER_OK;

  69.     if((ret = mutex_lock(&c->mutex)) != OPER_OK)    return ret;

  71.     /*
  72.      * Wait while the buffer is full.
  73.      */
  74.     while(cbuf_full(c))
  75.     {
  76. #ifdef DEBUG_CBUF
  77.     debug("cbuf is full !!!\n");
  78. #endif
  79.         cond_wait(&c->not_full,&c->mutex);
  80.     }

  82.     c->data[c->next_in++] = data;
  83.     c->size++;
  84.     c->next_in %= c->capacity;

  86.     mutex_unlock(&c->mutex);

  88.     /*
  89.      * Let a waiting consumer know there is data.
  90.      */
  91.     cond_signal(&c->not_empty);

  93. #ifdef DEBUG_CBUF
  94. //    debug("cbuf enqueue success ,data : %p\n",data);
  95.     debug("enqueue\n");
  96. #endif

  98.     return ret;
  99. }



  103. /* 取出数据 */
  104. void*        cbuf_dequeue(cbuf_t *c)
  105. {
  106.     void     *data     = NULL;
  107.     int32_t    ret     = OPER_OK;

  109.     if((ret = mutex_lock(&c->mutex)) != OPER_OK)    return NULL;

  111.        /*
  112.      * Wait while there is nothing in the buffer
  113.      */
  114.     while(cbuf_empty(c))
  115.     {
  116. #ifdef DEBUG_CBUF
  117.     debug("cbuf is empty!!!\n");
  118. #endif
  119.         cond_wait(&c->not_empty,&c->mutex);
  120.     }

  122.     data = c->data[c->next_out++];
  123.     c->size--;
  124.     c->next_out %= c->capacity;

  126.     mutex_unlock(&c->mutex);


  129.     /*
  130.      * Let a waiting producer know there is room.
  131.      * 取出了一个元素，又有空间来保存接下来需要存储的元素
  132.      */
  133.     cond_signal(&c->not_full);

  135. #ifdef DEBUG_CBUF
  136. //    debug("cbuf dequeue success ,data : %p\n",data);
  137.     debug("dequeue\n");
  138. #endif

  140.     return data;
  141. }


  144. /* 判断缓冲区是否为满 */
  145. bool        cbuf_full(cbuf_t    *c)
  146. {
  147.     return (c->size == c->capacity);
  148. }

  150. /* 判断缓冲区是否为空 */
  151. bool        cbuf_empty(cbuf_t *c)
  152. {
  153.     return (c->size == 0);
  154. }

  156. /* 获取缓冲区可存放的元素的总个数 */
  157. int32_t        cbuf_capacity(cbuf_t *c)
  158. {
  159.     return c->capacity;
  160. }



二，辅助文件
    为了提高程序的移植性，对线程相关进行封装。
    1，thread.h

点击(此处)折叠或打开

  1. #ifndef __THREAD_H__
  2. #define __THREAD_H__

  4. #ifdef __cplusplus
  5. extern "C" {
  6. #endif

  8. /* Define to prevent recursive inclusion 
  9. -------------------------------------*/
  10. #include "types.h"





  16. typedef    struct _mutex
  17. {
  18.     pthread_mutex_t        mutex;
  19. }mutex_t;


  22. typedef    struct _cond
  23. {
  24.     pthread_cond_t        cond;
  25. }cond_t;


  28. typedef    pthread_t        tid_t;
  29. typedef    pthread_attr_t    attr_t;
  30. typedef    void*    (* thread_fun_t)(void*);


  33. typedef    struct _thread
  34. {
  35.     tid_t            tid;
  36.     cond_t            *cv;
  37.     int32_t            state;
  38.     int32_t            stack_size;
  39.     attr_t         attr;
  40.     thread_fun_t    fun;
  41. }thread_t;



  45. /* mutex */
  46. extern    int32_t        mutex_init(mutex_t    *m);
  47. extern    int32_t        mutex_destroy(mutex_t    *m);
  48. extern    int32_t        mutex_lock(mutex_t    *m);
  49. extern    int32_t        mutex_unlock(mutex_t    *m);


  52. /* cond */
  53. extern    int32_t        cond_init(cond_t    *c);
  54. extern    int32_t        cond_destroy(cond_t    *c);
  55. extern    int32_t        cond_signal(cond_t *c);
  56. extern    int32_t        cond_wait(cond_t    *c,mutex_t *m);



  60. /* thread */
  61. /* 线程的创建，其属性的设置等都封装在里面 */
  62. extern    int32_t        thread_create(thread_t *t);
  63. //extern    int32_t        thread_init(thread_t    *t);

  65. #define    thread_join(t, p)     pthread_join(t, p)
  66. #define    thread_self()        pthread_self()
  67. #define    thread_sigmask        pthread_sigmask


  70. #ifdef __cplusplus
  71. }
  72. #endif

  74. #endif
  75. /* END OF FILE 
  76. ---------------------------------------------------------------*/


    2，thread.c

点击(此处)折叠或打开

  1. #include "thread.h"




  6. /* mutex */
  7. int32_t        mutex_init(mutex_t    *m)
  8. {
  9.     int32_t        ret = OPER_OK;

  11.     if((ret = pthread_mutex_init(&m->mutex, NULL)) != 0)
  12.         ret = -THREAD_MUTEX_INIT_ERROR;

  14.     return ret;
  15. }


  18. int32_t        mutex_destroy(mutex_t    *m)
  19. {
  20.     int32_t        ret = OPER_OK;

  22.     if((ret = pthread_mutex_destroy(&m->mutex)) != 0)
  23.         ret = -MUTEX_DESTROY_ERROR;

  25.     return ret;
  26. }



  30. int32_t        mutex_lock(mutex_t    *m)
  31. {
  32.     int32_t        ret = OPER_OK;

  34.     if((ret = pthread_mutex_lock(&m->mutex)) != 0)
  35.         ret = -THREAD_MUTEX_LOCK_ERROR;

  37.     return ret;
  38. }



  42. int32_t        mutex_unlock(mutex_t    *m)
  43. {
  44.     int32_t        ret = OPER_OK;

  46.     if((ret = pthread_mutex_unlock(&m->mutex)) != 0)
  47.         ret = -THREAD_MUTEX_UNLOCK_ERROR;
  48.     
  49.     return ret;
  50. }






  57. /* cond */
  58. int32_t        cond_init(cond_t    *c)
  59. {
  60.     int32_t        ret = OPER_OK;

  62.     if((ret = pthread_cond_init(&c->cond, NULL)) != 0)
  63.         ret = -THREAD_COND_INIT_ERROR;

  65.     return ret;
  66. }



  70. int32_t        cond_destroy(cond_t    *c)
  71. {
  72.     int32_t        ret = OPER_OK;

  74.     if((ret = pthread_cond_destroy(&c->cond)) != 0)
  75.         ret = -COND_DESTROY_ERROR;
  76.     
  77.     return ret;
  78. }



  82. int32_t        cond_signal(cond_t *c)
  83. {
  84.     int32_t        ret = OPER_OK;


  87.     if((ret = pthread_cond_signal(&c->cond)) != 0)
  88.         ret = -COND_SIGNAL_ERROR;
  89.     
  90.     return ret;
  91. }




  96. int32_t        cond_wait(cond_t    *c,mutex_t *m)
  97. {
  98.     int32_t        ret = OPER_OK;

  100.     if((ret = pthread_cond_wait(&c->cond, &m->mutex)) != 0)
  101.         ret = -COND_WAIT_ERROR;    
  102.     
  103.     return ret;
  104. }



三，测试
    1，测试代码

点击(此处)折叠或打开

  1. /* 
  2.  * cbuf begin
  3.  */
  4. #define        OVER    (-1)

  6. static        cbuf_t    cmd;
  7. static        int        line_1[200];
  8. static        int        line_2[200];
  9. //static        int        temp = 0;

  11. static        bool    line1_finish = false;
  12. static        bool    line2_finish = false;

  14. void*    producer_1(void *data)
  15. {
  16.     int32_t    i = 0;

  18.     for(i = 0; i < 200; i++)
  19.     {
  20.         line_1[i] = i+1000;
  21.         cbuf_enqueue(&cmd, &line_1[i]);

  23.         if(0 == (i % 9)) sleep(1);
  24.     }

  26.     line1_finish = true;

  28.     return NULL;
  29. }

  31. void*    producer_2(void *data)
  32. {
  33.     int32_t    i = 0;

  35.     for(i = 0; i < 200; i++)
  36.     {
  37.         line_2[i] = i+20000;
  38.         cbuf_enqueue(&cmd, &line_2[i]);

  40.         if(0 == (i % 9)) sleep(1);
  41.     }

  43.     line2_finish = true;

  45.     return NULL;
  46. }


  49. void*    consumer(void *data)
  50. {
  51.     int32_t        *ptr = NULL;

  53.     while(1)
  54.     {
  55.         ptr = cbuf_dequeue(&cmd);
  56.         printf("%d\n",*ptr);

  58.         if(cbuf_empty(&cmd) && line2_finish && line1_finish)
  59.         {
  60.             printf("quit\n");
  61.             break;
  62.         }
  63.     }

  65.     return NULL;
  66. }


  69. void    test_cbuf_oper(void)
  70. {
  71.     pthread_t    l_1;
  72.     pthread_t    l_2;
  73.     pthread_t    c;
  74.     
  75.     cbuf_init(&cmd);

  77.     pthread_create(&l_1,NULL,producer_1,0);
  78.     pthread_create(&l_2,NULL,producer_2,0);
  79.     pthread_create(&c,NULL,consumer,0);

  81.     pthread_join(l_1,NULL);
  82.     pthread_join(l_2,NULL);
  83.     pthread_join(c,NULL);

  85.     cbuf_destroy(&cmd);
  86. }


  89. void    test_cbuf(void)
  90. {
  91.     test_cbuf_oper();
  92. }


  95. /* 
  96.  * cbuf end
  97.  */


    2，测试结果



四，参考文件
1，《bareos-master》源码
2，《nginx》源码

文章标签：
Linux
C语言
关键词：
C语言缓冲区
Linux生产者
Linux消费者
Linux线程
Linux多线程
桃子红了呐
目录
相关文章
aliyun5518910231-27881
|
8月前
|
存储 Linux API
【Linux进程概念】—— 操作系统中的“生命体”，计算机里的“多线程”
在计算机系统的底层架构中，操作系统肩负着资源管理与任务调度的重任。当我们启动各类应用程序时，其背后复杂的运作机制便悄然展开。程序，作为静态的指令集合，如何在系统中实现动态执行？本文带你一探究竟！
aliyun5518910231-27881
306 6
【Linux进程概念】—— 操作系统中的“生命体”，计算机里的“多线程”
蓝易云
|
6月前
|
并行计算 Linux
Linux内核中的线程和进程实现详解
了解进程和线程如何工作，可以帮助我们更好地编写程序，充分利用多核CPU，实现并行计算，提高系统的响应速度和计算效能。记住，适当平衡进程和线程的使用，既要拥有独立空间的'兄弟'，也需要在'家庭'中分享和并行的成员。对于这个世界，现在，你应该有一个全新的认识。
蓝易云
266 67
hahaha无
|
4月前
|
存储 编译器 程序员
c语言的文件操作与文件缓冲区
如果没有文件，我们写的程序的数据是存储在电脑的内存中，如果程序退出，内存回收，数据就丢失了，等再次运行程序，是看不到上次程序的数据的，如果要将数据进行持久化的保存，我们可以使用文件。磁盘（硬盘）上的文件是文件。但是在程序设计中，我们⼀般谈的⽂件有两种：程序文件、数据文件（从文件功能的角度来分类 的）。就比如说我们电脑中以.txt为后缀的就是文件的一种，他就是数据文件。.exe为后缀的就为程序文件。函数名功能适用范围fgetc字符输入函数所有输入流fputc字符输出函数所有输出流fgets。
hahaha无
111 0
蓝易云
|
8月前
|
Linux
Linux编程： 在业务线程中注册和处理Linux信号
通过本文，您可以了解如何在业务线程中注册和处理Linux信号。正确处理信号可以提高程序的健壮性和稳定性。希望这些内容能帮助您更好地理解和应用Linux信号处理机制。
蓝易云
140 26
蓝易云
|
8月前
|
Linux
Linux编程： 在业务线程中注册和处理Linux信号
本文详细介绍了如何在Linux中通过在业务线程中注册和处理信号。我们讨论了信号的基本概念，并通过完整的代码示例展示了在业务线程中注册和处理信号的方法。通过正确地使用信号处理机制，可以提高程序的健壮性和响应能力。希望本文能帮助您更好地理解和应用Linux信号处理，提高开发效率和代码质量。
蓝易云
147 17
c的前世今生
|
12月前
|
存储 编译器 C语言
如何在 C 语言中判断文件缓冲区是否需要刷新？
在C语言中，可以通过检查文件流的内部状态或使用`fflush`函数尝试刷新缓冲区来判断文件缓冲区是否需要刷新。通常，当缓冲区满、遇到换行符或显式调用`fflush`时，缓冲区会自动刷新。
c的前世今生
241 4
c的前世今生
|
12月前
|
存储 编译器 C语言
C语言：文件缓冲区刷新方式有几种
C语言中文件缓冲区的刷新方式主要包括三种：自动刷新（如遇到换行符或缓冲区满）、显式调用 fflush() 函数强制刷新、以及关闭文件时自动刷新。这些方法确保数据及时写入文件。
c的前世今生
588 4
石小浪♪
|
资源调度 Linux 调度
Linux C/C++之线程基础
这篇文章详细介绍了Linux下C/C++线程的基本概念、创建和管理线程的方法，以及线程同步的各种机制，并通过实例代码展示了线程同步技术的应用。
石小浪♪
203 0
Linux C/C++之线程基础
AI小云
|
3月前
|
安全 算法 Java
Java 多线程：线程安全与同步控制的深度解析
本文介绍了 Java 多线程开发的关键技术，涵盖线程的创建与启动、线程安全问题及其解决方案，包括 synchronized 关键字、原子类和线程间通信机制。通过示例代码讲解了多线程编程中的常见问题与优化方法，帮助开发者提升程序性能与稳定性。
AI小云
157 0
兵临天下19970108016
|
3月前
|
数据采集 监控 调度
干货分享“用 多线程 爬取数据”：单线程 + 协程的效率反超 3 倍，这才是 Python 异步的正确打开方式
在 Python 爬虫中，多线程因 GIL 和切换开销效率低下，而协程通过用户态调度实现高并发，大幅提升爬取效率。本文详解协程原理、实战对比多线程性能，并提供最佳实践，助你掌握异步爬虫核心技术。
兵临天下19970108016
323 0

热门文章

最新文章

  • 1
    一、初识 Linux 与基本命令
  • 2
    Kali Linux 2025.3 发布 (Vagrant & Nexmon) - 领先的渗透测试发行版
  • 3
    Metasploit Framework 6.4.90 (macOS, Linux, Windows) - 开源渗透测试框架
  • 4
    七、Linux Shell 与脚本基础
  • 5
    二、Linux文本处理与文件操作核心命令
  • 6
    四、Linux核心工具：Vim, 文件链接与SSH
  • 7
    Tenable Nessus 10.10 (macOS, Linux, Windows) - 漏洞评估解决方案
  • 8
    六、Linux核心服务与包管理
  • 9
    Metasploit Pro 4.22.8-2025091701 (Linux, Windows) - 专业渗透测试框架
  • 10
    Binary Ninja 5.1.8104 (macOS, Linux, Windows) - 反编译器、反汇编器、调试器和二进制分析平台
  • 1
    万字详解：C语言三子棋进阶 + N子棋递归动态判断输赢（二）
    172
  • 2
    万字详解：C语言三子棋进阶 + N子棋递归动态判断输赢（一）
    239
  • 3
    超详解 - 如何理解C语言中while(scanf(“%d“, &num) != EOF)这一表达式？
    639
  • 4
    从C语言到C++_26(set+map+multiset+multimap)力扣692+349+牛客_单词识别（下）
    122
  • 5
    从C语言到C++_26(set+map+multiset+multimap)力扣692+349+牛客_单词识别（中）
    121
  • 6
    从C语言到C++_26(set+map+multiset+multimap)力扣692+349+牛客_单词识别（上）
    162
  • 7
    从C语言到C++_25(树的十道OJ题)力扣:606+102+107+236+426+105+106+144+94+145（下）
    116
  • 8
    从C语言到C++_25(树的十道OJ题)力扣:606+102+107+236+426+105+106+144+94+145（中）
    125
  • 9
    从C语言到C++_25(树的十道OJ题)力扣:606+102+107+236+426+105+106+144+94+145（上）
    99
  • 10
    从C语言到C++_24(二叉搜索树)概念+完整代码实现+笔试题（下）
    159

    • 相关课程

    更多
  • Linux高级网络应用 - 网络管理与配置实战
  • Linux Web服务器Nginx搭建与配置
  • Linux用户和组管理
  • Linux企业运维实战 - 入门及常用命令
  • Linux Shell 编程入门与实战
  • Linux网络进阶 - TCP/IP协议及OSI七层模型

    • 相关电子书

    更多
  • Alibaba Cloud Linux 3 发布
  • ECS系统指南之Linux系统诊断
  • ECS运维指南 之 Linux系统诊断

    • 推荐镜像

    更多
  • mxlinux-iso
  • archlinuxarm
  • archlinuxcn
    • 下一篇
    阿里云oss简介和如何对接使用