Linux Socket + pthread + pipe 实现socket通信和多线程数据共享

大道至简，基础的东西不能忘，对于网路时代，开发应该都跑不了socket，作为一个基本功，也好久没写写，简单了一个复习下。主要目标复习下socket的同时，复习下linux多线程处理
涉及知识点：
1，socket
2，linux pthread使用
3，信号量（semaphore）、锁（mutex）、管道（pipe）、条件
更多知识点，百度应该比较多了。
代码逻辑：
1，server：最大支持10个链接，当收到有效连接时启动一个线程用于处理对于事务（recvfromClient）
在线程中如果收到client发送字符串vote则触发另一个线程votepthread来通知（voteAction）policethread线程vote事件发生了（policeCenter）
2，client：相对简单就是支持循环输入
话不多说，代码如下：
server代码：

#include <iostream>
#include <string>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <pthread.h>
#include <unistd.h>
#include <semaphore.h>
using namespace std;
sem_t sem;
pthread_cond_t cond;
pthread_mutex_t mutex;
int pipeHandle[2];
void *voteAction(void *data) {
  while(1) {
    pthread_cond_wait(&cond, &mutex);
    pthread_t pid = pthread_self();
    cout <<"people "<< pid <<" vote action happen, please call 911" << endl;
    write(pipeHandle[1], &pid, sizeof(pthread_t));
    sem_post(&sem);
  }
  return NULL;
}

void *policeCenter(void* data) {
  while(1) {
    sem_wait(&sem);
    pthread_t senderId;
    read(pipeHandle[0],&senderId, sizeof(pthread_t));
    cout << "911 center recevice people "<< senderId <<" notify vote event happened" << endl;
  }
  return NULL;
}

void *recvfromClient(void *data) {
  int connfd = (int)(*((int *)data));
  char buff[1024];
  while (1) {
    memset(buff, 0, sizeof(buff));
    int ret = recv(connfd, buff, sizeof(buff), 0);
    if (ret == -1) {
      cout << "socket recv data fail" << endl;
      break;
    }
    if (string(buff) == "vote") {
      pthread_cond_signal(&cond);
    }
    cout << "server recv data:" << buff << endl;
  }
  close(connfd);
  pthread_exit(NULL);
  return NULL;
}

int main() {
  int socketfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); // 0 : 采用type默认的protocol
  if (socketfd == -1) {
    cout << "socket create fail" << endl;
    return -1;
  }

  struct sockaddr_in addr;
  memset(&addr,0,sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = htonl(INADDR_ANY);
  addr.sin_port = htons(6666);
  if (bind(socketfd, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
    cout << "socker bind fail" << endl;
    return -1;
  }

  if (listen(socketfd, 10) == -1) {
    cout << "socket listen fail" << endl;
    return -1;
  }

  // thread for vote
  pthread_t voteThread;
  int ret = pthread_create(&voteThread, NULL, voteAction, NULL);
  if (ret < 0) {
    cout << "create vote thread fail" << endl;
    close(socketfd);
    return -1;
  }
  pthread_cond_init(&cond, NULL);
  pthread_mutex_init(&mutex, NULL);

  pthread_t policeThread;
  ret = pthread_create(&policeThread, NULL, policeCenter, NULL);
  if (ret < 0) {
    cout << "create police center thread fail" << endl;
    //pthread_cance(voteThread);
    close(socketfd);
    return -1;
  }
  sem_init(&sem, 1, 0);

  //init pipe for share data between different threads
  ret = pipe(pipeHandle);
  if (ret < 0) {
    cout << "create pipe fail" << endl;
    close(socketfd);
    return -1;
  }

  cout << "server ready..." << endl;
  while(1) {
    int connfd = accept(socketfd, (struct sockaddr *)NULL, NULL);
    if (connfd == -1) {
      cout << "socket accept fail" << endl;
      continue;
    }
    pthread_t thread;
    int ret = pthread_create(&thread, NULL, recvfromClient, &connfd);
    if (ret < 0) {
      cout << "thread create fail" << endl;
      continue;
    }
    ret = pthread_detach(thread);
    if (ret < 0) {
      cout << "thread deatch fail" << endl;
    }
  }

  pthread_join(voteThread, NULL);
  pthread_join(policeThread, NULL);
  close(socketfd);
  return 0;
}

client代码如下：

#include <iostream>
#include <sys/socket.h>
#include <unistd.h>
#include <string.h>
#include <netinet/in.h>
using namespace std;
int main() {
  int socketfd = socket(AF_INET, SOCK_STREAM, 0);
  if (socketfd == -1) {
    cout << "socket create fail" << endl;
    return -1;
  }

  struct sockaddr_in addr;
  memset(&addr, 0, sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_port = htons(6666);
  int ret = connect(socketfd, (struct sockaddr*)&addr, sizeof(addr));
  if (ret < 0) {
    cout << "socket connect fail" << endl;
    close(socketfd);
    return -1;
  }

  while(1) {
    cout << "Please input the data:";
    string input;
    cin >> input;
    if (input.size() && send(socketfd, input.c_str(), input.size(), 0) < 0) {
      cout << "send msg fail" << endl;
    }
  }

  close(socketfd);
  return -1;
}

编译
g++ server.cpp -lpthread -o server
g++ client.cpp -lpthread -o client

 首先是用pipe在主线程和子线程通信，而且read是阻塞的

#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <fcntl.h>

typedef struct __Thread
{
    pthread_t tid;                 //thread id
    int notifyReceiveFd;           //the receive side of pipe
    int notifySendFd;              //the send side pf pipe
} Thread;

Thread* m_thread;

void* work_thread(void* argc)
{
    Thread* param=(Thread*) argc;
    printf("childthread_id=%lu\n",param->tid);
    int contant=0;
    printf("childthread--read return %ld\n",read(param->notifyReceiveFd, &contant, sizeof(int)));
    printf("childthread--read from pipe %d\n", contant);

    sleep(5);
    contant=100;
    //printf("childthread--write return %ld\n", write(param->notifySendFd, &contant, sizeof(int)));

}

int main(int argc, char** argv)
{
    //build pipe between main-thread and child-thread
    m_thread = new Thread;
    int fds[2];
    if(pipe(fds))
    {
        perror("create pipe fds_1 error\n");
    }

    int fds_1[2];
    if(pipe(fds_1))
    {
        perror("create pipe fds_1 error\n");
    }

    m_thread->notifyReceiveFd=fds[0];
    m_thread->notifySendFd=fds_1[1];

    pthread_t id;


    pthread_create(&(m_thread->tid),NULL,work_thread,(void*)m_thread);
    printf("mainthread--childthread id is %lu\n",m_thread->tid);

    int contant = 1;

    printf("mainthread--write %d to pipe\n",contant);
    printf("mainthread--write return %ld\n", write(fds[1], &contant, sizeof(int)));

    //下面的read()函数会阻塞，因为没有给fds_1管道写入数据（把work_thread中注释掉的write那行去掉注释，则read()就可以读到数据而不阻塞）
    printf("mainthread--read return %ld\n", read(fds_1[0], &contant, sizeof(int)));

    printf("mainthread--read from pipe %d\n", contant);
    pthread_join(m_thread->tid,NULL);
    close(fds[0]);
    close(fds[1]);
    close(fds_1[0]);
    close(fds_1[1]);
}

 上面的代码会被阻塞，若要不阻塞运行，则要把文件描述符设置为非阻塞的，则read会立即返回（当管道没有数据可读时返回-1）

    
    fcntl(fds_1[0], F_SETFL, O_NONBLOCK);
    //在read前，用fcnl设置fds_1[0]为非阻塞的
    printf("mainthread--read return %ld\n", read(fds_1[0], &contant, sizeof(int)));

如要设置超时,可以使用select或poll，下面代码只举例select。更多相关知识可以参考这篇文章

#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <fcntl.h>

typedef struct __Thread
{
    pthread_t tid;                 //thread id
    int notifyReceiveFd;           //the receive side of pipe
    int notifySendFd;              //the send side pf pipe
} Thread;

Thread* m_thread;

void* work_thread(void* argc)
{
    Thread* param=(Thread*) argc;
    printf("childthread_id=%lu\n",param->tid);
    int contant=0;
    printf("childthread--read return %ld\n",read(param->notifyReceiveFd, &contant, sizeof(int)));
    printf("childthread--read from pipe %d\n", contant);


    sleep(5);
    contant=100;
    //printf("childthread--write return %ld\n", write(param->notifySendFd, &contant, sizeof(int)));

}

int main(int argc, char** argv)
{
    //build pipe between main-thread and child-thread
    m_thread = new Thread;
    int fds[2];
    if(pipe(fds))
    {
        perror("create pipe fds_1 error\n");
    }

    int fds_1[2];
    if(pipe(fds_1))
    {
        perror("create pipe fds_1 error\n");
    }

    m_thread->notifyReceiveFd=fds[0];
    m_thread->notifySendFd=fds_1[1];

    pthread_t id;


    pthread_create(&(m_thread->tid),NULL,work_thread,(void*)m_thread);
    printf("mainthread--childthread id is %lu\n",m_thread->tid);

    int contant = 1;

    printf("mainthread--write %d to pipe\n",contant);
    printf("mainthread--write return %ld\n", write(fds[1], &contant, sizeof(int)));

    fd_set m_fds;
    FD_ZERO(&m_fds);
    FD_SET(fds_1[0], &m_fds);
    struct timeval t = {10, 0};
    select(fds_1[0] + 1, &m_fds, NULL, NULL, &t);
//若10后内fds_1[0]不可读则停止阻塞，程序继续往下走；如果把work_thread中的write那行去掉注释，
//则注释到write执行完毕，因为我write前sleep了5s，所以我这里会阻塞5s，然后程序继续往下走
    printf("mainthread--read return %ld\n", read(fds_1[0], &contant, sizeof(int)));

    printf("mainthread--read from pipe %d\n", contant);
    pthread_join(m_thread->tid,NULL);
    close(fds[0]);
    close(fds[1]);
    close(fds_1[0]);
    close(fds_1[1]);
}

在linux下进行多线程编程，肯定会涉及到线程通信问题，本文主要分析pipe，即管道在多线之间通信实现。
#include<unistd.h>
int pipe(int filedes[2]);
返回值：成功，返回0，否则返回-1。
参数数组包含pipe使用的两个文件的描述符。fd[0]:读管道，fd[1]:写管道

两个线程之间通信简单实现，单向pipe_1.c
源码地址：https://github.com/jeremy505/multi-thread-communication

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
Thread *m_Threads;
static int threadcount = 1;
void* work_thread(void* argc)
{
  Thread* param = (Thread*) argc;
  printf("childthread_tid=%lu\n", param->tid);
  int contant = 0;
  //sleep(2);
  printf("childthread--read return %d\n",read(param->notifyReceiveFd, &contant, sizeof(int)));
  printf("childthread--read from pipe %d\n", contant);
}

int main(int argc, char** argv)
{
     //在主线程和子线程之间建立管道
    m_Threads = malloc(sizeof(Thread) * threadcount);
    int fds[2];
    if( pipe(fds) )
    {
      perror("create pipe error");
    }
    m_Threads[0].notifyReceiveFd = fds[0];
    pthread_create(&m_Threads[0].tid, NULL,  work_thread, (void*)&m_Threads[0]);
    printf("mainthread_tid=%lu\n", m_Threads[0].tid);
    int contant = 1;
    // sleep(2);
    printf("mainthread--write %d to pipe\n", contant);
    printf("mainthread--write return %d\n",write(fds[1], &contant, sizeof(int)));
    pthread_join(m_Threads[0].tid, NULL);
    close(fds[0]);
    close(fds[1]);
    return 0;
}

以上只是单向通信，即主线程向子线程写int型值1，子线程读到int型值1.
输出见下：


可见，输出与设想的一致，只是此时通信只是单向，如果要实现双向，最简单的办法就是再创建一组pipe，实现pipe_1_d.c如下：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
typedef struct __Thread
{
  pthread_t tid;    //线程的ID
  int notifyReceiveFd;  //管道的接收端
  int notifySendFd;   //管道的发送端
}Thread;

Thread *m_Threads;
static int threadcount = 1;
void* work_thread(void* argc)
{
  Thread* param = (Thread*) argc;
  printf("childthread_tid=%lu\n", param->tid);
  int contant = 0;
  //sleep(2);
  printf("childthread--read return %d\n",read(param->notifyReceiveFd, &contant, sizeof(int)));
  printf("childthread--read from pipe %d\n", contant);
  contant = 2;
  printf("childthread--write %d to pipe\n", contant);
  printf("childthread--write return %d\n",write(param->notifySendFd, &contant, sizeof(int)));
}

int main(int argc, char** argv)
{
     //在主线程和子线程之间建立管道
    m_Threads = malloc(sizeof(Thread) * threadcount);
    int fds[2];
    if( pipe(fds) )
    {
      perror("create pipe fds error");
    }
    int fds_1[2];
    if( pipe(fds_1) )
    {
      perror("create pipe fds_1 error");
    }
    m_Threads[0].notifyReceiveFd = fds[0];
    m_Threads[0].notifySendFd = fds_1[1];
     pthread_create(&m_Threads[0].tid, NULL,
                   work_thread, (void*)&m_Threads[0]);
    printf("mainthread_tid=%lu\n", m_Threads[0].tid);
    int contant = 1;
    // sleep(2);
    printf("mainthread--write %d to pipe\n", contant);
    printf("mainthread--write return %d\n",write(fds[1], &contant, sizeof(int)));
    printf("mainthread--read return %d\n",read(fds_1[0], &contant, sizeof(int)));
    printf("mainthread--read from pipe %d\n", contant);
    pthread_join(m_Threads[0].tid, NULL);
    close(fds[0]);
    close(fds[1]);
    close(fds_1[0]);
    close(fds_1[1]);
    }

另外创建一组pipe，主线读，子线程写。主线程先写1，然后阻塞等待子线程网管道中写值，子线程通过管道读到1之后往管道写2，此时管道有数据，主线程读取值2，输出如下：

以上只是简单的通过pipe线程之间进行同行，注意到读写都是阻塞的。如果不希望线程使用阻塞方式，一般会设置管道文件描述符为非阻塞，然后借助epoll或者select监听管道文件描述符读写事件。
线程之间通过pipe通信也可以应用到进程之间，在使用fork之后，管道描述符被拷贝了一份，所以父子进程必须关闭其中之一，假设父进程关闭读[close(fd[0])]，子进程就要关闭写[close(fd[1])]，，实现单向通信，反过来也是一样.
进程之间的通信，推荐一种方式使用共享内存，共享内存区是最快的IPC形式，此种方式也可在两个完全独立的程序之间进行数据传递，后续再详细介绍。
当然，进程以及线程之间的通信不止以上方法，还有使用socket，eventfd等。