备案控制台
探索云世界
开发者社区 开发与运维 文章 正文

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

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

转自：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语言
关键词：
Linux线程
线程通信
Linux C语言
C语言Linux
Linux缓冲区
桃子红了呐
目录
相关文章
程序三两行
|
15天前
|
存储 Java 数据库连接
java多线程之线程通信
java多线程之线程通信
程序三两行
26 1
fensnote
|
22天前
|
Linux C++
LInux下Posix的传统线程示例
LInux下Posix的传统线程示例
fensnote
19 1
泡沫o0
|
29天前
|
算法 Unix Linux
Linux与Qt线程优先级的对应关系：一次全面解析
Linux与Qt线程优先级的对应关系：一次全面解析
泡沫o0
21 0
泡沫o0
|
30天前
|
消息中间件 Linux 调度
【Linux 进程/线程状态 】深入理解Linux C++中的进程/线程状态：阻塞，休眠，僵死
【Linux 进程/线程状态 】深入理解Linux C++中的进程/线程状态：阻塞，休眠，僵死
泡沫o0
67 0
一歲抬頭
|
1天前
|
固态存储 Ubuntu Linux
Linux(29) 多线程快速解压缩|删除|监视大型文件
Linux(29) 多线程快速解压缩|删除|监视大型文件
一歲抬頭
9 1
fensnote
|
22天前
|
小程序 Linux API
Linux用C语言模拟‘ls‘命令
Linux用C语言模拟‘ls‘命令
fensnote
11 1
凉开水白菜
|
23天前
|
Linux 测试技术 C语言
【Linux】应用编程之C语言文件操作
【Linux】应用编程之C语言文件操作
凉开水白菜
21 0
泡沫o0
|
28天前
|
监控 Linux 调度
【Linux 应用开发 】Linux 下应用层线程优先级管理解析
【Linux 应用开发 】Linux 下应用层线程优先级管理解析
泡沫o0
45 0
泡沫o0
|
29天前
|
存储 算法 Linux
【Linux 系统标准 进程资源】Linux 创建一个最基本的进程所需的资源分析,以及线程资源与之的差异
【Linux 系统标准 进程资源】Linux 创建一个最基本的进程所需的资源分析,以及线程资源与之的差异
泡沫o0
25 0
泡沫o0
|
1月前
|
存储 运维 Linux
【Shell 命令集合 系统设置 】⭐Linux 显示Linux内核环缓冲区的内容 dmesg命令 使用指南
【Shell 命令集合 系统设置 】⭐Linux 显示Linux内核环缓冲区的内容 dmesg命令 使用指南
泡沫o0
34 0
【Shell 命令集合 系统设置 】⭐Linux 显示Linux内核环缓冲区的内容 dmesg命令 使用指南

热门文章

最新文章

  • 1
    Java中的多线程编程：概念、实现与性能优化
  • 2
    【Qt 底层之事件驱动系统】深入理解 Qt 事件机制：主事件循环与工作线程的交互探究，包括 QML 的视角
  • 3
    【C/C++ 单线程性能分析工具 Gprof】 GNU的C/C++ 性能分析工具 Gprof 使用全面指南
  • 4
    Python 学习之路 01基础入门---【Python安装，Python程序基本组成】
  • 5
    【深入探究 Qt 线程】一文详细解析Qt线程的内部原理与实现策略
  • 6
    【软件设计师备考 专题 】操作系统的内核（中断控制）、进程、线程概念
  • 7
    【C/C++ 线程池设计思路】 深入探索线程池设计：任务历史记录的高效管理策略
  • 8
    【C++ const 函数 的使用】C++ 中 const 成员函数与线程安全性：原理、案例与最佳实践
  • 9
    线程同步与IPC：单进程多线程环境下的选择与权衡
  • 10
    深入理解Java并发编程：线程安全与性能优化
  • 1
    C 语言解 常见编程题（下）
    37
  • 2
    振南技术干货集：C语言的一些“骚操作”及其深层理解（2）
    13
  • 3
    【C语言】用三种循环语句 计算1到1000之间能被2或3整除的数的总和
    12
  • 4
    【C++对于C语言的扩充】C++与C语言的联系，命名空间、C++中的输入输出以及缺省参数
    25
  • 5
    爱上C语言：操作符详解（下）
    23
  • 6
    爱上C语言:扫雷小游戏,展开一片功能实现
    16
  • 7
    爱上C语言：分支与循环（分支篇）多个if与if — else if区别
    17
  • 8
    嵌入式C语言（六）
    19
  • 9
    C语言库函数 — 内存函数（含模拟实现内存函数）
    24
  • 10
    C语言库函数 — 字符串函数（含模拟实现字符串函数）
    35

    • 相关课程

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

    • 相关电子书

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

    • 相关实验场景

    更多
  • 使用计算巢企业应用，一键获取专属的Linux服务器管理软件
  • Alibaba Cloud Linux操作系统的安装及使用
  • Alibaba Cloud Linux操作系统Shell程序
  • 手动部署MySQL数据库（Alibaba Cloud Linux 2）
  • 部署并使用Docker（Alibaba Cloud Linux 3）
  • 部署并使用Docker（Alibaba Cloud Linux 2）

    • 推荐镜像

    更多
  • mxlinux-iso
  • archlinuxarm
  • archlinuxcn
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）