在之前的文章中（开发利器——C 语言必备实用第三方库），笔者介绍了一款Linux/UNIX下C语言库Melon的基本功能，并给出了一个简单的多进程开箱即用的例子。

本文将给大家介绍Melon中多线程的使用方法。

在Melon中有三种多线程模式：

• 多线程框架
• 线程池
• I/O线程

我们将逐一给出实例。

Melon的Github仓库：https://github.com/Water-Melon/Melon。

多线程框架

注意：Windows下目前不支持本功能。

多线程框架与前面介绍的线程池不同，是一种模块化线程。模块化线程是指，每一个线程都是一个独立的代码模块，都有各自对应的入口函数（类似于每一个 C 语言程序有一个 main 函数一样）。

自2.3.0版本起，多线程模块不再是以在Melon目录下的threads目录中编写代码文件的方式进行集成，而是采用注册函数，由使用者在程序中进行注册加载。这样可以解除线程模块代码与Melon库在目录结构上的耦合。

开发流程

1. 首先，编写一个源文件：

   #include <stdio.h>
   #include <assert.h>
   #include <string.h>
   #include <errno.h>
   #include "mln_framework.h"
   #include "mln_log.h"
   #include "mln_thread.h"static int haha(int argc, char **argv)
   {int fd = atoi(argv[argc-1]);mln_thread_msg_t msg;int nfds;fd_set rdset;for (;;) {FD_ZERO(&rdset);FD_SET(fd, &rdset);nfds = select(fd+1, &rdset, NULL, NULL, NULL);if (nfds < 0) {if (errno == EINTR) continue;mln_log(error, "select error. %s\n", strerror(errno));return -1;}memset(&msg, 0, sizeof(msg));
   #if defined(WIN32)int n = recv(fd, (char *)&msg, sizeof(msg), 0);
   #elseint n = recv(fd, &msg, sizeof(msg), 0);
   #endifif (n != sizeof(msg)) {mln_log(debug, "recv error. n=%d. %s\n", n, strerror(errno));return -1;}mln_log(debug, "!!!src:%S auto:%l char:%c\n", msg.src, msg.sauto, msg.c);mln_thread_clear_msg(&msg);}return 0;
   }static void hello_cleanup(void *data)
   {mln_log(debug, "@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@\n");
   }static int hello(int argc, char **argv)
   {mln_thread_cleanup_set(hello_cleanup, NULL);int i;for (i = 0; i < 1; ++i)  {int fd = atoi(argv[argc-1]);mln_thread_msg_t msg;memset(&msg, 0, sizeof(msg));msg.dest = mln_string_new("haha");assert(msg.dest);msg.sauto = 9736;msg.c = 'N';msg.type = ITC_REQUEST;msg.need_clear = 1;
   #if defined(WIN32)int n = send(fd, (char *)&msg, sizeof(msg), 0);
   #elseint n = send(fd, &msg, sizeof(msg), 0);
   #endifif (n != sizeof(msg)) {mln_log(debug, "send error. n=%d. %s\n", n, strerror(errno));mln_string_free(msg.dest);return -1;}}usleep(100000);return 0;
   }int main(int argc, char *argv[])
   {mln_thread_module_t modules[] = {{"haha", haha},{"hello", hello},};struct mln_framework_attr cattr;cattr.argc = argc;cattr.argv = argv;cattr.global_init = NULL;cattr.main_thread = NULL;cattr.worker_process = NULL;cattr.master_process = NULL;mln_thread_module_set(modules, 2);if (mln_framework_init(&cattr) < 0) {fprintf(stderr, "Melon init failed.\n");return -1;}return 0;
   }
   

   这段代码中有几个注意事项：

   • 在main中，我们使用mln_thread_module_set函数将每个线程模块的入口与线程模块名映射关系加载到Melon库中。
   • 在mln_framework_init会根据下面步骤中对配置文件的设置来自动初始化子线程，并进入线程开始运行。
   • 可以看到，两个线程模块之间是可以通信的，通信的方式是使用最后一个参数（argv的最后一个），来向主线程发送消息，消息中包含了目的线程的名字。这里注意，argv是在后面步骤中在配置文件中给出的，但是最后一个参数是程序自动在配置项后增加的一个参数，为主子线程间的通信套接字文件描述符。
   • 在 hello 这个模块中，调用了mln_thread_cleanup_set函数，这个函数的作用是：在从当前线程模块的入口函数返回至上层函数后，将会被调用，用于清理自定义资源。每一个线程模块的清理函数只能被设置一个，多次设置会被覆盖，清理函数是线程独立的，因此不会出现覆盖其他线程处理函数的情况（当然，你也可以故意这样来构造，比如传一个处理函数指针给别的模块，然后那个模块再进行设置）。

2. 对该文件进行编译：

   $ cc -o test test.c -I /path/to/melon/include -L /path/to/melon/lib -lmelon -lpthread
   

3. 修改配置文件

   log_level "none";
   //user "root";
   daemon off;
   core_file_size "unlimited";
   //max_nofile 1024;
   worker_proc 1;
   framework "multithread";
   log_path "/home/niklaus/melon/logs/melon.log";
   /** Configurations in the 'proc_exec' are the* processes which are customized by user.** Here is an example to show you how to* spawn a program.*     keepalive "/tmp/a.out" ["arg1" "arg2" ...]* The command in this example is 'keepalive' that* indicate master process to supervise this* process. If process is killed, master process* would restart this program.* If you don't want master to restart it, you can*     default "/tmp/a.out" ["arg1" "arg2" ...]** But you should know that there is another* arugment after the last argument you write here.* That is the file descriptor which is used to* communicate with master process.*/
   proc_exec {// keepalive "/tmp/a";
   }
   thread_exec {restart "hello" "hello" "world";default "haha";
   }
   

   这里主要关注framework以及thread_exec的配置项。thread_exec配置块专门用于模块化线程之用，其内部每一个配置项均为线程模块。

   以 hello 为例：

   restart "hello" "hello" "world";
   

   restart或者default是指令，restart表示线程退出主函数后，再次启动线程。而default则表示一旦退出便不再启动。其后的hello字符串就是模块的名称，其余则为模块参数，即入口函数的argc和argv的部分。而与主线程通信的套接字则不必写在此处，而是线程启动后进入入口函数前自动添加的。

4. 运行程序

   $ ./test
   Start up worker process No.1
   Start thread 'hello'
   Start thread 'haha'
   04/14/2022 14:50:16 UTC DEBUG: a.c:haha:34: PID:552165 !!!src:hello auto:9736 char:N
   04/14/2022 14:50:16 UTC DEBUG: a.c:hello_cleanup:42: PID:552165 @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
   04/14/2022 14:50:16 UTC REPORT: PID:552165 Thread 'hello' return 0.
   04/14/2022 14:50:16 UTC REPORT: PID:552165 Child thread 'hello' exit.
   04/14/2022 14:50:16 UTC REPORT: PID:552165 child thread pthread_join's exit code: 0
   04/14/2022 14:50:16 UTC DEBUG: a.c:haha:34: PID:552165 !!!src:hello auto:9736 char:N
   04/14/2022 14:50:16 UTC DEBUG: a.c:hello_cleanup:42: PID:552165 @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
   04/14/2022 14:50:16 UTC REPORT: PID:552165 Thread 'hello' return 0.
   04/14/2022 14:50:16 UTC REPORT: PID:552165 Child thread 'hello' exit.
   04/14/2022 14:50:16 UTC REPORT: PID:552165 child thread pthread_join's exit code: 0
   04/14/2022 14:50:16 UTC DEBUG: a.c:haha:34: PID:552165 !!!src:hello auto:9736 char:N
   04/14/2022 14:50:17 UTC DEBUG: a.c:hello_cleanup:42: PID:552165 @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
   04/14/2022 14:50:17 UTC REPORT: PID:552165 Thread 'hello' return 0.
   04/14/2022 14:50:17 UTC REPORT: PID:552165 Child thread 'hello' exit.
   04/14/2022 14:50:17 UTC REPORT: PID:552165 child thread pthread_join's exit code: 0
   ...
   

   可以看到，事实上 Melon 中会启动工作进程来拉起其子线程，而工作进程数量由worker_proc配置项控制，如果多于一个，则每个工作进程都会拉起一组haha和hello线程。此外，我们也看到，hello线程退出后，清理函数被调用。

高级使用

除了使用mln_thread_module_set注册线程模块外，还可以使用API动态添加和删除线程。这一功能可以使得线程可以根据外部需求进行动态部署和下线。

我们来看一个简单的例子：

#include <stdio.h>
#include <errno.h>
#include "mln_framework.h"
#include "mln_log.h"
#include "mln_thread.h"
#include <unistd.h>int sw = 0;
char name[] = "hello";
static void main_thread(mln_event_t *ev);static int hello(int argc, char *argv[])
{while (1) {printf("%d: Hello\n", getpid());usleep(10);}return 0;
}static void timer_handler(mln_event_t *ev, void *data)
{if (sw) {mln_string_t alias = mln_string("hello");mln_thread_kill(&alias);sw = !sw;mln_event_timer_set(ev, 1000, NULL, timer_handler);} else {main_thread(ev);}
}static void main_thread(mln_event_t *ev)
{char **argv = (char **)calloc(3, sizeof(char *));if (argv != NULL) {argv[0] = name;argv[1] = NULL;argv[2] = NULL;mln_thread_create(ev, "hello", THREAD_DEFAULT, hello, 1, argv);sw = !sw;mln_event_timer_set(ev, 1000, NULL, timer_handler);}
}int main(int argc, char *argv[])
{struct mln_framework_attr cattr;cattr.argc = argc;cattr.argv = argv;cattr.global_init = NULL;cattr.main_thread = main_thread;cattr.worker_process = NULL;cattr.master_process = NULL;if (mln_framework_init(&cattr) < 0) {fprintf(stderr, "Melon init failed.\n");return -1;}return 0;
}

这段代码中，我们使用main_thread这个回调来让worker进程的主线程增加一些初始化处理。

在main_thread中分配了一个指针数组，用来作为线程入口参数。并且利用mln_thread_create函数创建了一个名为hello的线程，线程的入口函数是hello，这个线程的类型是THREAD_DEFAULT（退出后不会重启）。然后设置了一个1秒的定时器。

每秒钟进入一次定时处理函数，函数中通过全局变量sw来杀掉和创建hello线程。

hello线程则是死循环输出hello字符串。

这里有一个点要注意：如果要使用mln_thread_kill杀掉子线程，则子线程内不能使用mln_log来打印日志，因为可能会导致日志函数锁无法释放而致使主线程死锁。

线程池

在Melon中支持两种多线程模式，线程池是其中一种，另一种请参见后续的多线程框架文章。

注意：在每个进程中仅允许存在一个线程池。

头文件

#include "mln_thread_pool.h"

模块名

thread_pool

函数

mln_thread_pool_run

int mln_thread_pool_run(struct mln_thread_pool_attr *tpattr);struct mln_thread_pool_attr {void                              *main_data;mln_thread_process                 child_process_handler;mln_thread_process                 main_process_handler;mln_thread_data_free                free_handler;mln_u64_t                          cond_timeout; /*ms*/mln_u32_t                          max;mln_u32_t                          concurrency;
};
typedef int  (*mln_thread_process)(void *);
typedef void (*mln_thread_data_free)(void *);

描述：创建并运行内存池。

线程池由主线程进行管理和做一部分处理后下发任务，子线程组则接受任务进行处理。

初始状态下，是不存在子线程的，当有任务需要下发时会自动创建子线程。当任务处理完后，子线程会延迟释放，避免频繁分配释放资源。

其中参数结构体的每个成员含义如下：

• main_data 为主线程的用户自定义数据。
• child_process_handler 每个子线程的处理函数，该函数有一个参数为主线程下发任务时给出的数据结构指针，返回值为0表示处理正常，非0表示处理异常，异常时会有日志输出。
• main_process_handler 主线程的处理函数，该函数有一个参数为main_data，返回值为0表示处理正常，非0表示处理异常，异常时会有日志输出。一般情况下，主线程处理函数不应随意自行返回，一旦返回代表线程池处理结束，线程池会被销毁。
• free_handler 为资源释放函数。其资源为主线程下发给子线程的数据结构指针所指向的内容。
• cond_timeout为闲置子线程回收定时器，单位为毫秒。当子线程无任务处理，且等待时间超过该定时器时长后，会自行退出。
• max线程池允许的最大子线程数量。
• concurrency用于pthread_setconcurrency设置并行级别参考值，但部分系统并为实现该功能，因此不应该过多依赖该值。在Linux下，该值设为零表示交由本系统实现自行确定并行度。

返回值：本函数返回值与主线程处理函数的返回值保持一致

mln_thread_pool_resource_add

int mln_thread_pool_resource_add(void *data);

描述：将资源data放入到资源池中。本函数仅应由主线程调用，用于主线程向子线程下发任务所用。

返回值：成功则返回0，否则返回非0

mln_thread_quit

void mln_thread_quit(void);

描述：本函数用于告知线程池，关闭并销毁线程池。

返回值：无

mln_thread_resource_info

void mln_thread_resource_info(struct mln_thread_pool_info *info);struct mln_thread_pool_info {mln_u32_t                          max_num;mln_u32_t                          idle_num;mln_u32_t                          cur_num;mln_size_t                         res_num;
};

描述：获取当前线程池信息。信息会写入参数结构体中，结构体每个参数含义如下：

• max_num：线程池最大子线程数量
• idle_num：当前闲置子线程数量
• cur_num：当前子线程数量（包含闲置和工作中的子线程）
• res_num：当前尚未被处理的资源数量

返回值：无

示例

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "mln_thread_pool.h"static int main_process_handler(void *data);
static int child_process_handler(void *data);
static void free_handler(void *data);int main(int argc, char *argv[])
{struct mln_thread_pool_attr tpattr;tpattr.main_data = NULL;tpattr.child_process_handler = child_process_handler;tpattr.main_process_handler = main_process_handler;tpattr.free_handler = free_handler;tpattr.cond_timeout = 10;tpattr.max = 10;tpattr.concurrency = 10;return mln_thread_pool_run(&tpattr);
}static int child_process_handler(void *data)
{printf("%s\n", (char *)data);return 0;
}static int main_process_handler(void *data)
{int n;char *text;while (1) {if ((text = (char *)malloc(16)) == NULL) {return -1;}n = snprintf(text, 15, "hello world");text[n] = 0;mln_thread_pool_resource_add(text);usleep(1000);}
}static void free_handler(void *data)
{free(data);
}

I/O线程

I/O线程算是一种另类线程池结构。但是这个组件主要用于图形界面类的应用。通常情况下，图形界面应用都会存在一个用户线程和一个I/O线程，这样当I/O处理时就不会无法响应用户的操作（如：点击）了。

视频介绍

头文件

#include "mln_iothread.h"

模块名

iothread

函数/宏

mln_iothread_init

int mln_iothread_init(mln_iothread_t *t, struct mln_iothread_attr *attr);struct mln_iothread_attr {mln_u32_t                   nthread; //几个I/O线程mln_iothread_entry_t        entry; //I/O线程入口函数void                       *args; //I/O线程入口参数mln_iothread_msg_process_t  handler; //消息处理函数
};typedef void *(*mln_iothread_entry_t)(void *); //线程入口
typedef void (*mln_iothread_msg_process_t)(mln_iothread_t *t, mln_iothread_ep_type_t from, mln_iothread_msg_t *msg);//消息处理函数

描述：依据attr对t进行初始化。

返回值：成功返回0，否则返回-1

mln_iothread_destroy

void mln_iothread_destroy(mln_iothread_t *t);

描述：销毁一个iothread实例。

返回值：无

mln_iothread_send

extern int mln_iothread_send(mln_iothread_t *t, mln_u32_t type, void *data, mln_iothread_ep_type_t to, int feedback);

描述：发送一个消息类型为type，消息数据为data的消息给to的一端，并根据feedback来确定是否阻塞等待反馈。

返回值：

• 0 - 成功
• -1 - 失败
• 1 - 发送缓冲区满

mln_iothread_recv

int mln_iothread_recv(mln_iothread_t *t, mln_iothread_ep_type_t from);

描述：从from的一端接收消息。接收后会调用初始化时设置好的消息处理函数，对消息进行处理。

返回值：已接收并处理的消息个数

mln_iothread_sockfd_get

 mln_iothread_iofd_get(p,t)

描述：从p所指代的mln_iothread_t结构中，根据t的值，获取I/O线程或用户线程的通信套接字。一般是为了将其加入到事件中。

返回值：套接字描述符

注意：套接字仅是用来通知对方线程（或线程组），另一端线程（或线程组）有消息发送过来，用户可以使用epoll、kqueue、select等事件机制进行监听。

mln_iothread_msg_hold

mln_iothread_msg_hold(m)

描述：将消息m持有，此时消息处理函数返回，该消息也不会被释放。主要用于流程较长的场景。该函数仅作用于feedback类型的消息，无需反馈的消息理论上也不需要持有，因为不在乎消息被处理的结果。

返回值：无

mln_iothread_msg_release

mln_iothread_msg_release(m)

描述：释放持有的消息。该消息应该是feedback类型消息，非该类型消息则可能导致执行流程异常。

返回值：无

mln_iothread_msg_type

mln_iothread_msg_type(m)

描述：获取消息的消息类型。

返回值：无符号整型

mln_iothread_msg_data

mln_iothread_msg_data(m)

描述：获取消息的用户自定义数据。

返回值：用户自定义数据结构指针

示例

#include "mln_iothread.h"
#include <string.h>
#include <stdio.h>
#include <errno.h>static void msg_handler(mln_iothread_t *t, mln_iothread_ep_type_t from, mln_iothread_msg_t *msg)
{mln_u32_t type = mln_iothread_msg_type(msg);printf("msg type: %u\n", type);
}static void *entry(void *args)
{int n;mln_iothread_t *t = (mln_iothread_t *)args;while (1) {n = mln_iothread_recv(t, user_thread);printf("recv %d message(s)\n", n);}return NULL;
}int main(void)
{int i, rc;mln_iothread_t t;struct mln_iothread_attr tattr;tattr.nthread = 1;tattr.entry = (mln_iothread_entry_t)entry;tattr.args = &t;tattr.handler = (mln_iothread_msg_process_t)msg_handler;if (mln_iothread_init(&t, &tattr) < 0) {fprintf(stderr, "iothread init failed\n");return -1;}for (i = 0; i < 1000000; ++i) {if ((rc = mln_iothread_send(&t, i, NULL, io_thread, 1)) < 0) {fprintf(stderr, "send failed\n");return -1;} else if (rc > 0)continue;}sleep(1);mln_iothread_destroy(&t);sleep(3);printf("DONE\n");return 0;
}