您现在的位置是:首页 >其他 >线程初探——生产者、消费者模式网站首页其他

线程初探——生产者、消费者模式

追風者0x8ffd98a8 2024-06-14 17:17:26
简介线程初探——生产者、消费者模式

线程初探——生产者、消费者模式

helllo,大家好,这里是追風者频道。今天我们来聊一聊生产者、消费者模式。后期需要采用该模式来进行协议栈抓包架构的搭建,通过多线程对数据包进行分任务处理。

生产者消费者模式结构以及问题

生产者消费者模式的结构:

  1. 生产者线程。
  2. 消费者线程。
  3. 缓存区(暂存区)。

生产者消费者模式的痛点:

  1. 线程同步,对缓存区的操作要进行同步(生产者无限生产,撑爆缓存区或抹除原始数据;消费者无限消费,获取到错误数据,导致程序出错)。
  2. 循环使用缓存区(当生产者生产的数据超出数组索引上限时,需要从0开始使用数组)。
  3. 可以线程安全访问的信号量计数器。

下图为生产者消费者模式模拟图:

生产者消费者模式

解决问题

我们可以通过计数器信号量来解决 线程同步问题,控制线程执行。通过记录索引,来控制缓存区的使用。

	#include <stdio.h>
#include "sys_plat.h"

static sys_sem_t sem;
static sys_mutex_t mutex;
static int count;

static char buffer[100];
static int write_index, read_index, total;
static sys_sem_t read_sem, write_sem;



void thread1_entry (void * arg) {
    for (int i = 0; i < 2 * sizeof(buffer); i++) {
        // 当计数信号量为 0 时,无限期等待。
        sys_sem_wait(read_sem, 0);
        char data = buffer[read_index++];

        if (read_index >= sizeof(buffer)) {
            read_index = 0;
        }

        sys_mutex_lock(total);
        total--;
        sys_mutex_unlock(total);

        sys_sem_notify(write_sem);
        
        plat_printf("thread1 read data=%d	 index=%d
", data, read_index);
        sys_sleep(2000);
    }
    
    
    plat_printf("thread1 count is %d
", count);
    while (1) {
        plat_printf("this is thread1: %s
", (char *)arg);
        sys_sleep(1000);
        sys_sem_notify(sem);
    }
}

void thread2_entry (void * arg) {
    sys_sleep(1000);
    for (int i = 0; i < 2 * sizeof(buffer); i++) {
        // 当计数信号量为 满缓存区 时,无限期等待。
        sys_sem_wait(write_sem, 0);
        buffer[write_index++] = i;

        if (write_index >= sizeof(buffer)) {
            write_index = 0;
        }

        sys_mutex_lock(total);
        total++;
        sys_mutex_unlock(total);
        
        plat_printf("thread2 write data=%d	 index=%d
", i, write_index);

        sys_sem_notify(read_sem);
    }

    plat_printf("thread2 count is %d
", count);
    while (1) {
        sys_sem_wait(sem, 0);
        plat_printf("this is thread2: %s
", (char *)arg);
    }
}

int main (void) {
    // 读取信号量计数器,从 0 开始。
    read_sem = sys_sem_create(0);
    // 写入信号量计数器,从 满 开始。
    write_sem = sys_sem_create(sizeof(buffer));

    sem = sys_sem_create(0);
    mutex = sys_mutex_create();

    sys_thread_create(thread1_entry, "A");
    sys_thread_create(thread2_entry, "B");
    
    printf("hello,network!");
    return 0;
}
风语者!平时喜欢研究各种技术,目前在从事后端开发工作,热爱生活、热爱工作。