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

简介:

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

操作系统:ubuntu10.04

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

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

点击(此处)折叠或打开

  1. #ifndef __CBUF_H__
  2. #define __CBUF_H__

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

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


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


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

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

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

  27. /* 取出数据 */
  28. extern    void*        cbuf_dequeue(cbuf_t *c);


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

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

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


  35. #ifdef __cplusplus
  36. }
  37. #endif

  38. #endif
  39. /END OF FILE 
  40. ---------------------------------------------------------------*/


    2,cbuf.c

点击(此处)折叠或打开

  1. #include "cbuf.h"



  2. /* 初始化环形缓冲区 */
  3. int32_t        cbuf_init(cbuf_t *c)
  4. {
  5.     int32_t    ret = OPER_OK;

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

  13.     if((ret = cond_init(&c->not_full)!= OPER_OK)    
  14.     {
  15. #ifdef DEBUG_CBUF
  16.     debug("cbuf init fail ! cond not full init fail !\n");
  17. #endif
  18.         mutex_destroy(&c->mutex);
  19.         return ret;
  20.     }

  21.     if((ret = cond_init(&c->not_empty)!= OPER_OK)
  22.     {
  23. #ifdef DEBUG_CBUF
  24.     debug("cbuf init fail ! cond not empty init fail !\n");
  25. #endif
  26.         cond_destroy(&c->not_full);
  27.         mutex_destroy(&c->mutex);
  28.         return ret;
  29.     }

  30.     c->size     = 0;
  31.     c->next_in    = 0;
  32.     c->next_out = 0;
  33.     c->capacity    = CBUF_MAX;

  34. #ifdef DEBUG_CBUF
  35.     debug("cbuf init success !\n");
  36. #endif

  37.     return ret;
  38. }


  39. /* 销毁环形缓冲区 */
  40. void        cbuf_destroy(cbuf_t    *c)
  41. {
  42.     cond_destroy(&c->not_empty);
  43.     cond_destroy(&c->not_full);
  44.     mutex_destroy(&c->mutex);

  45. #ifdef DEBUG_CBUF
  46.     debug("cbuf destroy success \n");
  47. #endif
  48. }



  49. /* 压入数据 */
  50. int32_t        cbuf_enqueue(cbuf_t *c,void *data)
  51. {
  52.     int32_t    ret = OPER_OK;

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

  54.     /*
  55.      * Wait while the buffer is full.
  56.      */
  57.     while(cbuf_full(c))
  58.     {
  59. #ifdef DEBUG_CBUF
  60.     debug("cbuf is full !!!\n");
  61. #endif
  62.         cond_wait(&c->not_full,&c->mutex);
  63.     }

  64.     c->data[c->next_in++= data;
  65.     c->size++;
  66.     c->next_in %= c->capacity;

  67.     mutex_unlock(&c->mutex);

  68.     /*
  69.      Let a waiting consumer know there is data.
  70.      */
  71.     cond_signal(&c->not_empty);

  72. #ifdef DEBUG_CBUF
  73. //    debug("cbuf enqueue success ,data : %p\n",data);
  74.     debug("enqueue\n");
  75. #endif

  76.     return ret;
  77. }



  78. /* 取出数据 */
  79. void*        cbuf_dequeue(cbuf_t *c)
  80. {
  81.     void     *data     NULL;
  82.     int32_t    ret     = OPER_OK;

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

  84.        /*
  85.      * Wait while there is nothing in the buffer
  86.      */
  87.     while(cbuf_empty(c))
  88.     {
  89. #ifdef DEBUG_CBUF
  90.     debug("cbuf is empty!!!\n");
  91. #endif
  92.         cond_wait(&c->not_empty,&c->mutex);
  93.     }

  94.     data = c->data[c->next_out++];
  95.     c->size--;
  96.     c->next_out %= c->capacity;

  97.     mutex_unlock(&c->mutex);


  98.     /*
  99.      Let a waiting producer know there is room.
  100.      * 取出了一个元素,又有空间来保存接下来需要存储的元素
  101.      */
  102.     cond_signal(&c->not_full);

  103. #ifdef DEBUG_CBUF
  104. //    debug("cbuf dequeue success ,data : %p\n",data);
  105.     debug("dequeue\n");
  106. #endif

  107.     return data;
  108. }


  109. /* 判断缓冲区是否为满 */
  110. bool        cbuf_full(cbuf_t    *c)
  111. {
  112.     return (c->size == c->capacity);
  113. }

  114. /* 判断缓冲区是否为空 */
  115. bool        cbuf_empty(cbuf_t *c)
  116. {
  117.     return (c->size == 0);
  118. }

  119. /* 获取缓冲区可存放的元素的总个数 */
  120. int32_t        cbuf_capacity(cbuf_t *c)
  121. {
  122.     return c->capacity;
  123. }



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

点击(此处)折叠或打开

  1. #ifndef __THREAD_H__
  2. #define __THREAD_H__

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

  6. /* Define to prevent recursive inclusion 
  7. -------------------------------------*/
  8. #include "types.h"





  9. typedef    struct _mutex
  10. {
  11.     pthread_mutex_t        mutex;
  12. }mutex_t;


  13. typedef    struct _cond
  14. {
  15.     pthread_cond_t        cond;
  16. }cond_t;


  17. typedef    pthread_t        tid_t;
  18. typedef    pthread_attr_t    attr_t;
  19. typedef    void*    (* thread_fun_t)(void*);


  20. typedef    struct _thread
  21. {
  22.     tid_t            tid;
  23.     cond_t            *cv;
  24.     int32_t            state;
  25.     int32_t            stack_size;
  26.     attr_t         attr;
  27.     thread_fun_t    fun;
  28. }thread_t;



  29. /* mutex */
  30. extern    int32_t        mutex_init(mutex_t    *m);
  31. extern    int32_t        mutex_destroy(mutex_t    *m);
  32. extern    int32_t        mutex_lock(mutex_t    *m);
  33. extern    int32_t        mutex_unlock(mutex_t    *m);


  34. /* cond */
  35. extern    int32_t        cond_init(cond_t    *c);
  36. extern    int32_t        cond_destroy(cond_t    *c);
  37. extern    int32_t        cond_signal(cond_t *c);
  38. extern    int32_t        cond_wait(cond_t    *c,mutex_t *m);



  39. /* thread */
  40. /* 线程的创建,其属性的设置等都封装在里面 */
  41. extern    int32_t        thread_create(thread_t *t);
  42. //extern    int32_t        thread_init(thread_t    *t);

  43. #define    thread_join(t, p)     pthread_join(t, p)
  44. #define    thread_self()        pthread_self()
  45. #define    thread_sigmask        pthread_sigmask


  46. #ifdef __cplusplus
  47. }
  48. #endif

  49. #endif
  50. /END OF FILE 
  51. ---------------------------------------------------------------*/


    2,thread.c

点击(此处)折叠或打开

  1. #include "thread.h"




  2. /* mutex */
  3. int32_t        mutex_init(mutex_t    *m)
  4. {
  5.     int32_t        ret = OPER_OK;

  6.     if((ret = pthread_mutex_init(&m->mutexNULL)!= 0)
  7.         ret -THREAD_MUTEX_INIT_ERROR;

  8.     return ret;
  9. }


  10. int32_t        mutex_destroy(mutex_t    *m)
  11. {
  12.     int32_t        ret = OPER_OK;

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

  15.     return ret;
  16. }



  17. int32_t        mutex_lock(mutex_t    *m)
  18. {
  19.     int32_t        ret = OPER_OK;

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

  22.     return ret;
  23. }



  24. int32_t        mutex_unlock(mutex_t    *m)
  25. {
  26.     int32_t        ret = OPER_OK;

  27.     if((ret = pthread_mutex_unlock(&m->mutex)!= 0)
  28.         ret -THREAD_MUTEX_UNLOCK_ERROR;
  29.     
  30.     return ret;
  31. }






  32. /* cond */
  33. int32_t        cond_init(cond_t    *c)
  34. {
  35.     int32_t        ret = OPER_OK;

  36.     if((ret = pthread_cond_init(&c->condNULL)!= 0)
  37.         ret -THREAD_COND_INIT_ERROR;

  38.     return ret;
  39. }



  40. int32_t        cond_destroy(cond_t    *c)
  41. {
  42.     int32_t        ret = OPER_OK;

  43.     if((ret = pthread_cond_destroy(&c->cond)!= 0)
  44.         ret -COND_DESTROY_ERROR;
  45.     
  46.     return ret;
  47. }



  48. int32_t        cond_signal(cond_t *c)
  49. {
  50.     int32_t        ret = OPER_OK;


  51.     if((ret = pthread_cond_signal(&c->cond)!= 0)
  52.         ret -COND_SIGNAL_ERROR;
  53.     
  54.     return ret;
  55. }




  56. int32_t        cond_wait(cond_t    *c,mutex_t *m)
  57. {
  58.     int32_t        ret = OPER_OK;

  59.     if((ret = pthread_cond_wait(&c->cond&m->mutex)!= 0)
  60.         ret -COND_WAIT_ERROR;    
  61.     
  62.     return ret;
  63. }



三,测试
    1,测试代码

点击(此处)折叠或打开

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

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

  9. static        bool    line1_finish false;
  10. static        bool    line2_finish false;

  11. void*    producer_1(void *data)
  12. {
  13.     int32_t    i = 0;

  14.     for(i = 0; i < 200; i++)
  15.     {
  16.         line_1[i= i+1000;
  17.         cbuf_enqueue(&cmd&line_1[i]);

  18.         if(0 =(i % 9)) sleep(1);
  19.     }

  20.     line1_finish true;

  21.     return NULL;
  22. }

  23. void*    producer_2(void *data)
  24. {
  25.     int32_t    i = 0;

  26.     for(i = 0; i < 200; i++)
  27.     {
  28.         line_2[i= i+20000;
  29.         cbuf_enqueue(&cmd&line_2[i]);

  30.         if(0 =(i % 9)) sleep(1);
  31.     }

  32.     line2_finish true;

  33.     return NULL;
  34. }


  35. void*    consumer(void *data)
  36. {
  37.     int32_t        *ptr NULL;

  38.     while(1)
  39.     {
  40.         ptr = cbuf_dequeue(&cmd);
  41.         printf("%d\n",*ptr);

  42.         if(cbuf_empty(&cmd&& line2_finish && line1_finish)
  43.         {
  44.             printf("quit\n");
  45.             break;
  46.         }
  47.     }

  48.     return NULL;
  49. }


  50. void    test_cbuf_oper(void)
  51. {
  52.     pthread_t    l_1;
  53.     pthread_t    l_2;
  54.     pthread_t    c;
  55.     
  56.     cbuf_init(&cmd);

  57.     pthread_create(&l_1,NULL,producer_1,0);
  58.     pthread_create(&l_2,NULL,producer_2,0);
  59.     pthread_create(&c,NULL,consumer,0);

  60.     pthread_join(l_1,NULL);
  61.     pthread_join(l_2,NULL);
  62.     pthread_join(c,NULL);

  63.     cbuf_destroy(&cmd);
  64. }


  65. void    test_cbuf(void)
  66. {
  67.     test_cbuf_oper();
  68. }


  69. /
  70.  * cbuf end
  71.  */


    2,测试结果



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

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

热门文章

最新文章