? 兒子：“。。。。。。”

? 媽媽：“說話呀！壓歲錢呢？”

? 上面的對話就反映了邊緣觸發(fā)的工作模式，即輸入緩沖收到數(shù)據(jù)時僅注冊1次該事件，即使輸入緩沖中還留有數(shù)據(jù)，也不會再進行注冊。

? epoll默認以條件觸發(fā)方式工作，select也同樣是以該方式工作的，這兩種觸發(fā)方式各有利弊，使用哪種模式取決于特定的應(yīng)用場景和對性能的需求。邊緣觸發(fā)通常用于高性能服務(wù)器，因為它可以減少系統(tǒng)調(diào)用次數(shù)，而條件觸發(fā)則適用于簡單應(yīng)用或者對性能要求不高的場景。

3 線程的使用及線程同步

? 利用線程的高并發(fā)特點使客戶端的I/O操作分離，這樣可以使整個代碼更加清晰，并且能夠提高效率。

? 線程具有單獨的執(zhí)行流，因此需要單獨定義線程的main函數(shù)，還需要請求操作系統(tǒng)在單獨的執(zhí)行流中執(zhí)行該函數(shù)，完成該功能的函數(shù)如下：

#include<pthread.h>

int pthread_create(pthread_t * restrict thread, const pthread_attr_t * restrict attr, void * (* start_routine)(void *), void * restrict arg);
//成功時返回0，失敗時返回其他值
//thread：保存新創(chuàng)建進程ID的變量地址值
//attr：用于傳遞線程屬性的參數(shù)，傳遞NULL時，創(chuàng)建默認屬性的線程
//start_routine：相當(dāng)于線程main函數(shù)的、在單獨執(zhí)行流中執(zhí)行的函數(shù)地址值
//arg：通過第三個參數(shù)傳遞調(diào)用函數(shù)時包含傳遞參數(shù)信息的變量地址值

? 線程相關(guān)代碼在編譯時需要添加-lpthread選項聲明需要連接線程庫，只有這樣才能調(diào)用頭文件pthread.h中聲明的函數(shù)

? 由于線程的執(zhí)行是非阻塞的，所以如果不使用相關(guān)函數(shù)控制線程的執(zhí)行流，必須使用sleep函數(shù)阻塞進程，讓進程等待所有線程都返回之后在返回！

? 下面介紹控制線程執(zhí)行流的函數(shù)：

#include<pthread.h>

int pthread_join(pthread_t thread, void ** status);
//成功時返回0，失敗時返回其他值
//ptread：該參數(shù)值ID的線程終止后才會從該函數(shù)返回
//status：保存線程的main函數(shù)返回值的指針變量地址值

? 該函數(shù)可以阻塞進程的執(zhí)行。

? 當(dāng)多個線程同時運行時，就會產(chǎn)生臨界區(qū)的問題。

? 線程安全函數(shù)被多個線程同時調(diào)用時也不會引發(fā)問題，而非線程安全函數(shù)被同時調(diào)用時會引發(fā)問題。

? 在編譯時添加-D_REENTRANT宏使函數(shù)庫中的非線程安全函數(shù)變?yōu)槎x好的線程安全函數(shù)。

? 線程同步是在編寫有關(guān)線程程序時的核心問題，因為線程的調(diào)用順序是隨機的，并且同一個進程的線程共用一個棧區(qū)，所以共享變量的使用需要格外謹慎，修改共享變量的代碼我們稱之為臨界區(qū)。對于臨界區(qū)的訪問需要通過同步和互斥來進行保護：

• 同步：需要指定訪問同一內(nèi)存空間的線程執(zhí)行順序的情況

• 互斥：同時訪問同一內(nèi)存空間時發(fā)生的情況

? 互斥的實質(zhì)就是上鎖機制，其依賴互斥量來實現(xiàn)，下面介紹互斥量的創(chuàng)建及銷毀函數(shù)：

#include<pthread.h>

int pthread_mutex_init(pthread_mutex_t * mutex, const pthread_mutexattr * attr);
int pthread_mutex_destroy(pthread_mutex_t * mutex);
//成功時返回0，失敗時返回其他值
//mutex：創(chuàng)建互斥量時傳遞保存互斥量的變量地址值，銷毀時傳遞需要銷毀的互斥量地址值
//attr：傳遞即將創(chuàng)建的互斥量屬性，沒有特別需要指定的屬性時傳遞NULL

? 接下來介紹利用互斥量鎖住或釋放臨界區(qū)時使用的函數(shù)：

int pthread_mutex_lock(pthread_mutex_t * mutex);
int pthread_mutex_unlock(pthread_mutex_t * mutex);
//成功時返回0，失敗時返回其他值

? 函數(shù)用法：

pthread_mutex_lock(&mutex);
//臨界區(qū)的開始
//....
//臨界區(qū)的結(jié)束
pthread_mutex_unlock(&mutex);

? 為了提高效率，我們應(yīng)該最大限度減少互斥量lock、unlock函數(shù)的調(diào)用次數(shù)。

? 同步需要使用信號量(semaphore)來完成，下面介紹信號量創(chuàng)建及銷毀的方法。

#include<semaphore.h>

int sem_init(sem_t * sem, int pshared, unsigned int value);
int sem_destroy(sem_t * sem);
//成功時返回0，失敗時返回其他值
//sem：傳遞信號量時傳遞保存信號量的變量地址，銷毀時傳遞需要銷毀的信號變量地址值
//pshared：傳遞其他值時，創(chuàng)建可由多個進程共享的信號量；傳遞0時，創(chuàng)建只允許1個進程內(nèi)部使用的信號量
//value：指定新創(chuàng)建的信號量初始值

? 信號量中相當(dāng)于互斥量lock、unlock的函數(shù)，其中post讓信號量增加1，wait讓信號量減少1。

int sem_post(sem_t * sem);
int sem_wait(sem_t * sem);
//成功時返0，失敗時返回其他值

二、代碼實現(xiàn)

? 使用epoll實現(xiàn)聊天服務(wù)器的代碼不是很復(fù)雜，因為最核心的功能——數(shù)據(jù)的收發(fā)已經(jīng)被epoll函數(shù)很好地實現(xiàn)了，我們所要做的就是調(diào)用該函數(shù)并且在此基礎(chǔ)上添加一些簡單的功能。

chat_clnt.c:

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <pthread.h>

#define BUF_SIZE 100
#define NAME_SIZE 20

void *send_msg(void *arg);
void *recv_msg(void *arg);
void error_handling(char *msg);

char name[NAME_SIZE] = "[DEFAULT]";
char msg[BUF_SIZE];

int main(int argc, char *argv[])
{
	int sock;
	struct sockaddr_in serv_addr;
	pthread_t snd_thread, rcv_thread;
	void *thread_return;
	if (argc != 4)
	{
		printf("Usage : %s <IP> <port> <name>\n", argv[0]);
		exit(1);
	}

	sprintf(name, "[%s]", argv[3]);
	sock = socket(PF_INET, SOCK_STREAM, 0);

	memset(&serv_addr, 0, sizeof(serv_addr));
	serv_addr.sin_family = AF_INET;
	serv_addr.sin_addr.s_addr = inet_addr(argv[1]);
	serv_addr.sin_port = htons(atoi(argv[2]));

	if (connect(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) == -1)
		error_handling("connect() error");

	pthread_create(&snd_thread, NULL, send_msg, (void *)&sock);
	pthread_create(&rcv_thread, NULL, recv_msg, (void *)&sock);
	pthread_join(snd_thread, &thread_return);
	pthread_join(rcv_thread, &thread_return);
	close(sock);
	return 0;
}

void *send_msg(void *arg) // send thread main
{
	int sock = *((int *)arg);
	char name_msg[NAME_SIZE + BUF_SIZE];
	while (1)
	{
		fgets(msg, BUF_SIZE, stdin);
		if (!strcmp(msg, "q\n") || !strcmp(msg, "Q\n"))
		{
			close(sock);
			exit(0);
		}
		sprintf(name_msg, "%s %s", name, msg);
		write(sock, name_msg, strlen(name_msg));
	}
	return NULL;
}

void *recv_msg(void *arg) // read thread main
{
	int sock = *((int *)arg);
	char name_msg[NAME_SIZE + BUF_SIZE];
	int str_len;
	while (1)
	{
		str_len = read(sock, name_msg, NAME_SIZE + BUF_SIZE - 1);
		if (str_len == -1)
			return (void *)-1;
		name_msg[str_len] = 0;
		fputs(name_msg, stdout);
	}
	return NULL;
}

void error_handling(char *msg)
{
	fputs(msg, stderr);
	fputc('\n', stderr);
	exit(1);
}

clnt_serv.c:

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <pthread.h>

#define MAX_CLNT 256
#define BUF_SIZE 100
#define EPOLL_SIZE 50
void setnonblockingmode(int fd);
void error_handling(char *buf);

int main(int argc, char *argv[])
{
	int serv_sock, clnt_sock;
	struct sockaddr_in serv_adr, clnt_adr;
	socklen_t adr_sz;
	int str_len, i, j;
	char buf[BUF_SIZE];

	struct epoll_event *ep_events;
	struct epoll_event event;
	int epfd, event_cnt;

	int clnt_socks[MAX_CLNT];
	int clnt_cnt = 0;
	pthread_mutex_t mutex;

	if (argc != 2)
	{
		printf("Usage : %s <port>\n", argv[0]);
		exit(1);
	}

	serv_sock = socket(PF_INET, SOCK_STREAM, 0);
	memset(&serv_adr, 0, sizeof(serv_adr));
	serv_adr.sin_family = AF_INET;
	serv_adr.sin_addr.s_addr = htonl(INADDR_ANY);
	serv_adr.sin_port = htons(atoi(argv[1]));

	if (bind(serv_sock, (struct sockaddr *)&serv_adr, sizeof(serv_adr)) == -1)
		error_handling("bind() error");
	if (listen(serv_sock, 5) == -1)
		error_handling("listen() error");

	epfd = epoll_create(EPOLL_SIZE);
	ep_events = malloc(sizeof(struct epoll_event) * EPOLL_SIZE);

	setnonblockingmode(serv_sock);
	event.events = EPOLLIN;
	event.data.fd = serv_sock;
	epoll_ctl(epfd, EPOLL_CTL_ADD, serv_sock, &event);

	while (1)
	{
		event_cnt = epoll_wait(epfd, ep_events, EPOLL_SIZE, -1);
		if (event_cnt == -1)
		{
			puts("epoll_wait() error");
			break;
		}

		// puts("return epoll_wait");
		for (i = 0; i < event_cnt; i++)
		{
			if (ep_events[i].data.fd == serv_sock)//client comes!
			{
				adr_sz = sizeof(clnt_adr);
				clnt_sock = accept(serv_sock, (struct sockaddr *)&clnt_adr, &adr_sz);
				setnonblockingmode(clnt_sock);
				event.events = EPOLLIN | EPOLLET;
				event.data.fd = clnt_sock;
				pthread_mutex_lock(&mutex);
				clnt_socks[clnt_cnt++] = clnt_sock;
				pthread_mutex_unlock(&mutex);

				epoll_ctl(epfd, EPOLL_CTL_ADD, clnt_sock, &event);
				printf("connected client: %d \n", clnt_sock);
			}
			else
			{
				while (1)
				{
					str_len = read(ep_events[i].data.fd, buf, BUF_SIZE);
					if (str_len == 0) // close request!
					{
						epoll_ctl(epfd, EPOLL_CTL_DEL, ep_events[i].data.fd, NULL);
						close(ep_events[i].data.fd);
						pthread_mutex_lock(&mutex);
						for (j = 0; j < clnt_cnt; j++)
						{
							if (ep_events[i].data.fd == clnt_socks[j])
							{
								while (j < clnt_cnt - 1)
								{
									clnt_socks[j] = clnt_socks[j + 1];
									j++;
								}
								break;
							}
						}
						clnt_cnt--;
						pthread_mutex_unlock(&mutex);
						printf("closed client: %d \n", ep_events[i].data.fd);
						break;
					}
					else if (str_len < 0)
					{
						if (errno == EAGAIN)
							break;
					}
					else
					{
						pthread_mutex_lock(&mutex);
						for (j = 0; j < clnt_cnt; j++)
						{
							if (clnt_socks[j] != ep_events[i].data.fd)
								write(clnt_socks[j], buf, str_len);
						}
						pthread_mutex_unlock(&mutex);
					}
				}
			}
		}
	}
	close(serv_sock);
	close(epfd);
	return 0;
}

void setnonblockingmode(int fd)
{
	int flag = fcntl(fd, F_GETFL, 0);
	fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}
void error_handling(char *buf)
{
	fputs(buf, stderr);
	fputc('\n', stderr);
	exit(1);
}

Makefile：

CC = gcc
CFLAGS = -Wall -pthread
TARGET_SERVER = chat_serv
TARGET_CLIENT = chat_clnt

all: $(TARGET_SERVER) $(TARGET_CLIENT)

$(TARGET_SERVER): chat_serv.o
	$(CC) $(CFLAGS) -o $(TARGET_SERVER) chat_serv.o

$(TARGET_CLIENT): chat_clnt.o
	$(CC) $(CFLAGS) -o $(TARGET_CLIENT) chat_clnt.o

chat_serv.o: chat_serv.c
	$(CC) $(CFLAGS) -c chat_serv.c

chat_clnt.o: chat_clnt.c
	$(CC) $(CFLAGS) -c chat_clnt.c

clean:
	rm -f *.o $(TARGET_SERVER) $(TARGET_CLIENT)

三、運行結(jié)果展示

當(dāng)有客戶接入聊天室的時候會服務(wù)器會顯示其占用的文件描述符；每一個客戶在發(fā)送消息是會在其發(fā)送的消息前附加上其昵稱，這樣更容易識別是哪一個客戶發(fā)來的消息。文章來源地址http://www.zghlxwxcb.cn/news/detail-773867.html

到了這里，關(guān)于epoll并發(fā)服務(wù)器的實現(xiàn)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！