原文地址：http://blog.chinaunix.net/uid-24830931-id-3786670.html


一、现象描述


在利用librdkafka同kafka broker通信过程中，当kafka broker意外退出时（如kill -9），librdkafka接口的sendmsg接口报出了“Program received signal SIGPIPE, Broken pipe.” 这个错误具有典型性，根据网络搜索的结果，这个一般是由于向一个被破坏的socket连接或者pipe读写数据造成的，向有经验的同事请教，他们说这种场景不会出现SIGPIPE信号，而是直接send， write， sendmsg等返回-1，同时errno会被设置成EPIPE。

实践是检验真理的唯一标准，找个例子一试便知。

二、例子程序

为了快速检验，从网上上借了一个简单的客户端、服务器程序， http://hi.baidu.com/dlpucat/item/97ab75c5243b8761f6c95d75 ，多谢原作者。

服务器端程序 server.c

点击(此处)折叠或打开

  1. #include <netinet/in.h>
  2. #include <sys/types.h>
  3. #include <sys/socket.h>
  4. #include <stdio.h>
  5. #include <stdlib.h>
  6. #include <string.h>

  8. #define HELLO_WORLD_SERVER_PORT 6666
  9. #define LENGTH_OF_LISTEN_QUEUE 20
  10. #define BUFFER_SIZE 1024

  12. int main(int argc, char **argv)
  13. {
  14.   struct sockaddr_in server_addr;
  15.   bzero(&server_addr,sizeof(server_addr));
  16.   server_addr.sin_family = AF_INET;
  17.   server_addr.sin_addr.s_addr = htons(INADDR_ANY);
  18.   server_addr.sin_port = htons(HELLO_WORLD_SERVER_PORT);

  20.   int server_socket = socket(AF_INET,SOCK_STREAM,0);
  21.   if( server_socket < 0)
  22.   {
  23.     printf("Create Socket Failed!");
  24.     exit(1);
  25.   }

  27.   if( bind(server_socket,(struct sockaddr*)&server_addr,sizeof(server_addr)))
  28.   {
  29.     printf("Server Bind Port : %d Failed!", HELLO_WORLD_SERVER_PORT);
  30.     exit(1);
  31.   }

  33.   if ( listen(server_socket, LENGTH_OF_LISTEN_QUEUE) )
  34.   {
  35.     printf("Server Listen Failed!");
  36.     exit(1);
  37.   }

  39.   while (1)
  40.   {
  41.     struct sockaddr_in client_addr;
  42.     socklen_t length = sizeof(client_addr);

  44.     int new_server_socket = accept(server_socket,(struct sockaddr*)&client_addr,&length);
  45.     if ( new_server_socket < 0)
  46.     {
  47.       printf("Server Accept Failed!\n");
  48.       break;
  49.     }

  51.     char buffer[BUFFER_SIZE];
  52.     bzero(buffer, BUFFER_SIZE);
  53.     strcpy(buffer,"Hello,World from server!");
  54.     strcat(buffer,"\n");
  55.     send(new_server_socket,buffer,BUFFER_SIZE,0);

  57.     bzero(buffer,BUFFER_SIZE);
  58.         while(1){
  59.       length = recv(new_server_socket,buffer,BUFFER_SIZE,0);
  60.       if (length < 0)
  61.       {
  62.         printf("Server Recieve Data Failed!\n");
  63.         exit(1);
  64.       }
  65.       printf("\n%s",buffer);
  66.         }
  67.     close(new_server_socket);
  68.   }
  69.   close(server_socket);
  70.   return 0;
  71. }

客户端程序

点击(此处)折叠或打开

  1. #include <netinet/in.h>
  2. #include <sys/types.h>
  3. #include <sys/socket.h>
  4. #include <stdio.h>
  5. #include <stdlib.h>
  6. #include <string.h>
  7. #include <signal.h>
  8. #include <errno.h>

  10. #define HELLO_WORLD_SERVER_PORT 6666
  11. #define BUFFER_SIZE 1024

  13. int main(int argc, char **argv)
  14. {
  15.   if (argc != 2)
  16.   {
  17.     printf("Usage: ./%s ServerIPAddress\n",argv[0]);
  18.     exit(1);
  19.   }

  21.   struct sockaddr_in client_addr;
  22.   bzero(&client_addr,sizeof(client_addr));
  23.   client_addr.sin_family = AF_INET;
  24.   client_addr.sin_addr.s_addr = htons(INADDR_ANY);
  25.   client_addr.sin_port = htons(0);

  27.   int client_socket = socket(AF_INET,SOCK_STREAM,0);

  29.   if( client_socket < 0)
  30.   {
  31.     printf("Create Socket Failed!\n");
  32.     exit(1);
  33.   }

  35.   if( bind(client_socket,(struct sockaddr*)&client_addr,sizeof(client_addr)))
  36.   {
  37.     printf("Client Bind Port Failed!\n");
  38.     exit(1);
  39.   }

  41.   struct sockaddr_in server_addr;
  42.   bzero(&server_addr,sizeof(server_addr));
  43.   server_addr.sin_family = AF_INET;
  44.   if(inet_aton(argv[1],&server_addr.sin_addr) == 0)
  45.   {
  46.     printf("Server IP Address Error!\n");
  47.     exit(1);
  48.   }
  49.   server_addr.sin_port = htons(HELLO_WORLD_SERVER_PORT);
  50.   socklen_t server_addr_length = sizeof(server_addr);
  51.   if(connect(client_socket,(struct sockaddr*)&server_addr, server_addr_length) < 0)
  52.   {
  53.     printf("Can Not Connect To %s!\n",argv[1]);
  54.     exit(1);
  55.   }

  57.   char buffer[BUFFER_SIZE];
  58.   bzero(buffer,BUFFER_SIZE);
  59.   int length = recv(client_socket,buffer,BUFFER_SIZE,0);
  60.   if(length < 0)
  61.   {
  62.     printf("Recieve Data From Server %s Failed!\n", argv[1]);
  63.     exit(1);
  64.   }
  65.   printf("From Server %s :\t%s",argv[1],buffer);

  67.   bzero(buffer,BUFFER_SIZE);
  68.   strcpy(buffer,"Hello, World! From Client\n");

  70.   while(1){
  71.     sleep(1);
  72.     int ret = send(client_socket,buffer,BUFFER_SIZE,0);
  73.         if (ret == -1 && errno == EPIPE){
  74.       printf("receive sigpipe\n");
  75.     }
  76.   }

  78.   close(client_socket);
  79.   return 0;
  80. }

三、重现方法

step 1）编译： gcc -o server server.c
          gcc -o -g client client.c （通过gdb直接看到异常退出）

step 2）启动服务器端：./server

step 3) 启动客户端： （这里假设客户端和服务器部署在同一台服务器 ） gdb ./client 
(gdb) r 127.0.0.1

step 4) 观察正常运行结果：首先是客户端收到服务器端的消息：From Server 127.0.0.1 : Hello,World from server!
         然后是服务器端每隔1s收到客户端的消息： Hello, World! From Client

step 5）通过ctrl+c关闭服务器端

step 6）观察客户端结果
Program received signal SIGPIPE, Broken pipe.
0x0000003a7fcd55f5 in send () from /lib64/libc.so.6

重现了！！

四、解决办法

解决办法很多，也很简单。

4.1 client中忽略SIGPIPE信号

点击(此处)折叠或打开

  1. signal(SIGPIPE, SIG_IGN);

4.2 阻止SIGPIPE信号（后来追查，原来同事的程序框架中已经有了这种机制，所以没有经历过程序退出的问题）

点击(此处)折叠或打开

  1. sigset_t set;
  2. sigemptyset(&set);
  3. sigaddset(&set, SIGPIPE);
  4. sigprocmask(SIG_BLOCK, &set, NULL);

4.3  为SIGPIPE添加信号处理函数，处理完程序继续执行

点击(此处)折叠或打开

  1. signal(SIGPIPE, pipesig_handler);

多种选择，总有一款适合您

# linux
Logo
Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南