li_wen01

li_wen01

关注
发信
关注(28)粉丝(399)

一种基于Unix Domain和TCP连接的跨设备多进程间通信的方法

前言：

   在linux系统进程间通信的方式有消息，消息队列，管道，内存映射，套接字等多种方式。在Android系统上进行进程间通信主要是使用Binder，其它的还有共享内存，管道，RPC和Unix Domain等方式。但是，在linux中常用的消息队列，在Android等系统上并不能直接的使用，Android上常用的Binder，在其他的系统上同样不能使用，如果要在windows,linux,android这样的不同平台上实现同一套进程间命令消息通信机制，并且有较好的移植性，那么在进行进程间通信设计的时候，首先应该考虑socket方式，这样方便以后设备功能的扩展。

Unix domain基础:

  1. 在使用套接字进行网络连接的时候，我们常用的也就是大家熟悉的是TCP/UDP连接，它属于AF_INET(IPV4)或是AF_INET6(IPV6)地址家族，在linux系统，socket还包括其他的一些地址家族：AF_UNIX,AF_IPX,AF_NETLINK,AF_X25,AF_AX25,AF_ATMPVC,AF_APPLETALK,AF_PACKET,AF_ALG。        
  2. 这里我们会使用到AF_UNIX地址家族，也就是Unix Domain Socket。可以确认的是在linux，Android，Mac和window10系统都支持AF_UNIX地址家族，这方便我们代码的移植复用。AF_UNIX类似于管道，依赖路径名标识发送方和接收方，从而实现本地进程间通信。即发送数据时，指定接收方绑定的路径名，操作系统根据该路径名可以直接找到对应的接收方，并将原始数据直接拷贝到接收方的内核缓冲区中，并上报给接收方进程进行处理。同样的接收方可以从收到的数据包中获取到发送方的路径名，并通过此路径名向其发送数据。
  3. AF_INET(TCP/UDP)需经过多个协议层的编解码，消耗系统cpu，并且数据传输需要经过网卡，受到网卡带宽的限制。AF_UNIX数据到达内核缓冲区后，由内核根据指定路径名找到接收方socket对应的内核缓冲区，直接将数据拷贝过去，不经过协议层编解码，节省系统cpu，并且不经过网卡，因此不受网卡带宽的限制。
  4. AF_UNIX的传输速率远远大于AF_INET。

设计思路：

(1)协议选择

    由于Unix Domain传输速率大消耗资源少，但它只适用于本地之间传输，AF_INET Domain 消耗资源多，传输相对速率较低。为了高效实现跨设备多进程间通讯，可以同时使用Unix Domain 和 AF_INET Domain进程系统进程间通信。本地进程使用Unix Domain传输,远程设备使用AF_INET传输，为保证命令消息的可靠发送和接收，可以选择TCP传输。

(2)多进程间通信实现

    要现实某个进程往其他任意一个进程间发送数据，中间必须建立一个服务端，服务端用来做数据路由，用来减少网络端口的连接同时简化各进程间数据的接收和发送。另外还可以通过路由实现广播的功能。

(3)命令消息格式设计

    在实际应用中，多进程间的通信一般是用来实现各进程间的命令消息交互，为了保证命令消息的正确和完整，一般会对命令消息进行封装，这样也方便接收端在接收到数据的时候进行命令的解析。该命令消息的格式可以定义如下：

一种基于Unix Domain和TCP连接的跨设备多进程间通信的方法-LMLPHP

基本的通信网络模型如下：

一种基于Unix Domain和TCP连接的跨设备多进程间通信的方法-LMLPHP

功能实现:

(1)服务端(Router)

(a)与客户端建立映射

    服务端与每一个客户端之间，只建立了一个连接。那么服务端要怎么知道当前是哪个客户端与服务端建立连接的呢？服务端又是如何知道要将接收到的数据发到对应的那个网络连接中去呢？这里我们需要为每一个客户端分配一个固定的地址，服务端通过这个地址来判断是哪个客户端请求建立连接。

    对于TCP连接，可以通过IP和端口来确定绑定地址，对于Unix Domain客户端，可以通过路径名来绑定地址。地址定义和绑定如下,这里预定义了20本地进程(客户端)，两个远程设备，每个设备10个进程(客户端)：

/************************************************************
*Copyright (C),lcb0281at163.com lcb0281atgmail.com
*FileName: ipc_common.h
*BlogAddr: https://blog.csdn.net/li_wen01
*Description:
	socket进程间通信数据结构及参数定义
	包括跨设备间TCP进程间通信和本地Unix Domain
	Socket进程间通信
*Date:	   2019-08-03
*Author:   Caibiao Lee
*Version:  V1.0
*Others:
*History:
***********************************************************/
#ifndef _IPC_COMMON_H_
#define _IPC_COMMON_H_

/**跨设备TCP socket进程间通信**/
#define MAX_EXTARN_DEV_NUM			  2      /**最大连接的网络设备数**/
#define EACH_TCP_DEV_MAX_CLIENT_NUM   10     /**每个外接网络设备客户端数**/
#define TCP_SERVER_PORT				  6666   /**TCP连接服务端IP端口**/

/**TCP连接最大支持客户端数**/
#define TCP_SERVER_LISTEN_MAX_NUM	 ((MAX_EXTARN_DEV_NUM)*(EACH_TCP_DEV_MAX_CLIENT_NUM))

#define TCP_SERVER_IP				"192.168.1.111"
#define TCP_CLIENT_DEVICE1_IP		"192.168.1.111"  /**网络设备1 IP地址**/
#define TCP_CLIENT_DEVICE2_IP		"192.168.1.112"  /**网络设备2 IP地址**/
#define TCP_CLIENT_DEVICE1_NO		1
#define TCP_CLIENT_DEVICE2_NO		2

/**网络设备各客户端TCP端口**/
#define TCP_DEVICE_CLIENT0_POART	9000
#define TCP_DEVICE_CLIENT1_POART	9001
#define TCP_DEVICE_CLIENT2_POART	9002
#define TCP_DEVICE_CLIENT3_POART	9003
#define TCP_DEVICE_CLIENT4_POART	9004
#define TCP_DEVICE_CLIENT5_POART	9005
#define TCP_DEVICE_CLIENT6_POART	9006
#define TCP_DEVICE_CLIENT7_POART	9007
#define TCP_DEVICE_CLIENT8_POART	9008
#define TCP_DEVICE_CLIENT9_POART	9009
#define TCP_DEVICE_CLIENT_MAX_POART 9009


/**本地Unix Domain Socket进程间通信**/
#define MAX_UDS_CLIENT_NUM			  20        /**最大socket域客户端数，对应最大本地通讯进程数**/
#define	SERVER_PATH		"../tmp/server_socket"   /**socket域服务端文件**/
#define CLIENT_PACHT    "../tmp/client_socket"   /**socket域客户端文件前缀**/

/**本地客户端模块序号定义**/
#define CLIENT_MIN_ADDR					1
#define LOCAL_CLIENT_1_ADDR        		1
#define LOCAL_CLIENT_2_ADDR        		2
#define LOCAL_CLIENT_3_ADDR        		3
#define LOCAL_CLIENT_4_ADDR        		4
#define LOCAL_CLIENT_5_ADDR        		5
#define LOCAL_CLIENT_6_ADDR        		6
#define LOCAL_CLIENT_7_ADDR        		7
#define LOCAL_CLIENT_8_ADDR        		8
#define LOCAL_CLIENT_9_ADDR        		9
#define LOCAL_CLIENT_10_ADDR        	10
#define LOCAL_CLIENT_11_ADDR        	11
#define LOCAL_CLIENT_12_ADDR        	12
#define LOCAL_CLIENT_13_ADDR        	13
#define LOCAL_CLIENT_14_ADDR        	14
#define LOCAL_CLIENT_15_ADDR        	15
#define LOCAL_CLIENT_16_ADDR        	16
#define LOCAL_CLIENT_17_ADDR        	17
#define LOCAL_CLIENT_18_ADDR        	18
#define LOCAL_CLIENT_19_ADDR        	19
#define LOCAL_CLIENT_MAX_ADDR        	19


/**网络设备1客户端模块序号定义**/
#define DEV_CLIENT_MIN_ADDR        		20
#define DEV1_CLIENT_0_ADDR        		20
#define DEV1_CLIENT_1_ADDR        		21
#define DEV1_CLIENT_2_ADDR        		22
#define DEV1_CLIENT_3_ADDR        		23
#define DEV1_CLIENT_4_ADDR        		24
#define DEV1_CLIENT_5_ADDR        		25
#define DEV1_CLIENT_6_ADDR        		26
#define DEV1_CLIENT_7_ADDR        		27
#define DEV1_CLIENT_8_ADDR        		28
#define DEV1_CLIENT_9_ADDR        		29

/**网络设备1客户端模块序号定义**/
#define DEV2_CLIENT_0_ADDR        		30
#define DEV2_CLIENT_1_ADDR        		31
#define DEV2_CLIENT_2_ADDR        		32
#define DEV2_CLIENT_3_ADDR        		33
#define DEV2_CLIENT_4_ADDR        		34
#define DEV2_CLIENT_5_ADDR        		35
#define DEV2_CLIENT_6_ADDR        		36
#define DEV2_CLIENT_7_ADDR        		37
#define DEV2_CLIENT_8_ADDR        		38
#define DEV2_CLIENT_9_ADDR        		39
#define DEV_CLIENT_MAX_ADDR        		39
#define CLIENT_MAX_ADDR      			39

extern const char * gc_au8DeviceNoMap[MAX_EXTARN_DEV_NUM];
extern const unsigned int gc_as32DeviceModuleMap[TCP_SERVER_LISTEN_MAX_NUM][3];

(b)客户端与socket连接ID映射

    服务端(Router)程序起来之后，建立两个线程，分别去监听TCP和Unix Domain客户端的连接请求。与客户端建立连接后，将相应的socket ID保存到一个数组中去，将数组地址与客户端地址设计成相同，这样在服务端发送数据的时候，可以通过数组地址直接快速查找到对应客户端的socket ID 。通过直接地址映射会比遍历数组会快很多，可以节省系统开销。

(c)数据路由

    服务端接收到数据之后，直接获取该命令消息的目的地址，然后再将该命令消息包发送到对应的进程。在使用socket转发数据的时候，需要注意，在socket接收一次数据，可能不是一个完整的数据包，也有可能该数据包里面有多条命令消息，这里在做数据处理的时候需要特别注意。    

    服务端server.c实现代码比较长，这里就不贴出来了，有兴趣的可以到我GitHub上查看：https://github.com/licaibiao/IPC_Socket

(2)客户端(各进程)

    为方便客户端对消息命令的处理，客户端设计成每次接收到的数据是一个完整的命令消息，并且只有一条命令消息。为方便客户端的使用，将客户端的网络连接和网络状态检测同时封装到客户端数据接收和数据发送两个接口里面。比如在数据发送的时候，会去判断连接是否建立获取连接已经断开，如果网络异常则重新建立连接。

客户端ipc_interface.h接口定义如下

/************************************************************
*Copyright (C),lcb0281at163.com lcb0281atgmail.com
*FileName: ipc_interface.c
*BlogAddr: https://blog.csdn.net/li_wen01
*Description:客户端网络连接，数据收发，消息解析函数接口定义实现
*Date:	   2019-08-03
*Author:   Caibiao Lee
*Version:  V1.0
*Others:
*History:
***********************************************************/
#ifndef _IPC_INTERFACE_H_
#define _IPC_INTERFACE_H_

#include "ipc_common.h"
#include "ipc_msgstruct.h"


/**获取流水号**/
int IPCP_Arch_Msg_AnalyzeGetFlowNum();

/**获取应答流水号**/
int IPCP_Arch_Msg_AnalyzeGetACKResult(ARCH_MSG_S *pstMsg);

/**获取流水号地址**/
int IPCP_Arch_Msg_AnalyzeGetRecFlow(ARCH_MSG_S *pstMsg);

/**获取源地址**/
int IPCP_Arch_Msg_ChangeSrcAddr(ARCH_MSG_S *pstMsg,unsigned  char SrcAddr);

/**获取目标地址**/
int IPCP_Arch_Msg_ChangeTargAddr(ARCH_MSG_S *pstMsg,unsigned  char TargAddr);

/**获取源地址**/
int IPCP_Arch_Msg_AnalyzeGetSrcAddr(ARCH_MSG_S *pstMsg);

/**获取目标地址**/
int IPCP_Arch_Msg_AnalyzeGetTargAddr(ARCH_MSG_S *pstMsg);

/**获取消息ID**/
int IPCP_Arch_Msg_AnalyzeGetCmdID(ARCH_MSG_S *pstMsg);

/**获取内容长度**/
int IPCP_Arch_Msg_AnalyzeGetLen(ARCH_MSG_S *pstMsg);

/**获取消息内容开始位置**/
void IPCP_Arch_Msg_PlatformStartP(unsigned char ** p, unsigned char* Data);

/**发送数据**/
int IPCP_Arch_Msg_PackSend(int s32ModuleAddr, MSG_PACK_S *pstMsg);

/**读取数据**/
int  IPCP_Arch_Msg_Recv(int s32ModuleAddr,ARCH_MSG_S *pstMsg);


#endif

(3)命令消息封装

    为了更好地发送和接收命令消息，应该将命令消息基于网络传输协议之上再进行一层消息的封装，添加消息头标签，源地址目的地址和校验等信息。为了各进程间消息的更好识别和传输，应该对每条命令的数据结构进行定义，这样在解析的时候才不会出现参数对应不上的问题。我这里预定义了几个命令消息，用来测试该方法的稳定性。

ipc_msgstruct.h

/************************************************************
*Copyright (C),lcb0281at163.com lcb0281atgmail.com
*FileName: ipc_msgstruct.h
*BlogAddr: https://blog.csdn.net/li_wen01
*Description:命令消息结构体定义和解析
*Date:	   2019-08-03
*Author:   Caibiao Lee
*Version:  V1.0
*Others:
*History:
***********************************************************/
#ifndef _IPC_MSGSTRUCT_H_
#define _IPC_MSGSTRUCT_H_

/**数据结构重定义**/
typedef unsigned char    byte;
typedef signed char	     INT8S;
typedef signed int	     INT32S;
typedef unsigned char    BYTE;
typedef unsigned short   WORD;
typedef unsigned int     DWORD;
typedef unsigned char    INT8U;
typedef unsigned short	 INT16U;
typedef unsigned int     INT32U;


#define IPCP_TRUE 			0
#define IPCP_FALSE			1

#define ARRAY_SIZE(_A) (sizeof(_A) / sizeof((_A)[0]))

/**通讯协议位定义**/
/************************************************************************************/
/** | 2Byet	| Byet	 | Byet	   | 2Byet  | 2*Byet |2Byet    |N*Byet | Byet | 2Byet  | **/
/** | 0xa5a5 | 源地址| 目标地址| 流水号 | 消息ID |消息长度 |消息体 | 校验 | 0x5a5a | **/
/** | 0  1   |   2   |    3    |  4  5  |  6  7  | 8  9    |10	   | 10+N | 10+N+1 | **/
/************************************************************************************/
#define MSG_SRC_ADDR 				(2)
#define MSG_TAR_ADDR 				(3)
#define MSG_SERIAL_NUM_ADDR 		(4)
#define MSG_CMID_ADDR 				(6)
#define MSG_MSG_LEN_ADDR 			(8)
#define MSG_ACK_CMID_ADDR 			(10)
#define MSG_ACK_SERIAL_NUM_ADDR 	(12)
#define MSG_ACK_RES_ADDR 			(14)
#define MSG_CONTENT_OFFSET 	 		(10)    /**消息内容开始位置**/

#define MSG_HEAD_LEN 				(10)
#define MSG_END_LEN  				(3)
#define MSG_ARCH_MSG_LEN  			(13)

#define QUEUE_MSG_HEAD              0xa5a5
#define QUEUE_MSG_END               0x5a5a


typedef struct MsgPack
{
    unsigned char SrcAddr;
    unsigned char TargAddr;
    unsigned short Len;
    unsigned char *Data;
    unsigned short CmdId;
}MSG_PACK_S;


#define  MSG_TEXT_SIZE       (2048)

typedef struct ARCH_MSG
{
    unsigned int MsgLen;
    unsigned char SomeText[MSG_TEXT_SIZE];
}ARCH_MSG_S;


/*通用IPCP指令*/
#define QUEUE_DEBUG_CMD1       		 (0x0001)
#define QUEUE_DEBUG_CMD2       		 (0x0002)
#define QUEUE_DEBUG_CMD3       		 (0x0003)
#define QUEUE_DEBUG_CMD4       		 (0x0004)
#define QUEUE_DEBUG_CMD5       		 (0x0005)
#define QUEUE_DEBUG_CMD6       		 (0x0006)
#define QUEUE_DEBUG_CMD7       		 (0x0007)
#define QUEUE_DEBUG_CMD8       		 (0x0008)
#define QUEUE_DEBUG_CMD9       		 (0x0009)
#define QUEUE_DEBUG_CMDA       		 (0x000A)
#define QUEUE_DEBUG_MAX_CMD			 (0x000A)


/**消息定义**/
typedef struct
{
	DWORD   u32Alarm;      		   /* 报警 */
	DWORD 	u32Status;     		   /* 状态 */
	DWORD 	u32Latitude;   		   /* 纬度，百万分之一度 */
	DWORD 	u32Longtitude; 		   /* 经度，百万分之一度 */
	WORD	u16Altitude;		   /* 高程，米 */
	WORD	u16SpeedX10;   		   /* gps速度，1/10 km/h */
	WORD  	u16Direct;     		   /* 方向 */
 	BYTE    arrCardNo[16];	       /* 机动车牌号码			 */
	WORD    u16SensorSpeed;        /* 脉冲速度，1/10 km/h */
	BYTE    u8CarCor;			   /* 车牌颜色 	 */
}__attribute__ ((__packed__))MSG_0X0001_S;

typedef struct
{
	unsigned int  u32FileId;	    /**文件id，不为0时，按文件id查找**/
	unsigned char u8DeleteFlag;		/**删除标志:0：保留；1：删除**/
	unsigned char u8StopFlag;		/**停止上传标志:0：保留；1：停止上传	**/
	unsigned char u8srcPlat;	    /**下发命令的平台地址**/
}__attribute__ ((__packed__))MSG_0X0002_S;

typedef struct
{
	unsigned char u8Type;		     /**0人脸识别结果，1人头个数识别**/
	unsigned char u8Result;          /**type为0时，0成功，1失败**/
}__attribute__ ((__packed__))MSG_0X0003_S;

typedef struct
{
	unsigned char u8Interval;		/**时间间隔**/
	unsigned int  u32Duration;    	/**持续时间**/
}__attribute__ ((__packed__))MSG_0X0004_S;

typedef struct
{
    DWORD u32MultiId;				/**多媒体ID,小端内存**/
   	WORD  u16TotalNum;				/**下发拍照的总数**/
    WORD  u16CurNum;				/**当前图片序号**/
	BYTE  u8PlatAddr;				/**平台地址**/
}__attribute__ ((__packed__))MSG_0X0005_S;

typedef struct
{
	DWORD   u32ModeEvent;      		/**休眠唤醒事件:1：休眠，2：唤醒*/
}__attribute__ ((__packed__))MSG_0X0006_S;

typedef struct
{
	DWORD    u32ParaId;   		   /* 参数id: 0x1000010C*/
	BYTE     u8ParaLen;			   /* 参数长度: 4 */
	unsigned long long u64AlarmFlag;/* bit*/

}__attribute__ ((__packed__))MSG_0X0007_S;

typedef struct
{
	DWORD    u32ParaId;   		  /* 参数id: 0x1000010C*/
	BYTE     u8ParaLen;			  /* 参数长度: 4 */
	DWORD    u32TimeOut;		  /* 单位为秒*/
}__attribute__ ((__packed__))MSG_0X0008_S;

typedef struct
{
	DWORD    u32ParaId;   		  /* 参数id: 0x30000007*/
	BYTE     u8ParaLen;			  /* 参数长度: 4 */
	DWORD    u32Payload;		  /* 参数值:96:h264编码, 265:h265编码*/
}__attribute__ ((__packed__))MSG_0X0009_S;

typedef struct
{
	BYTE	u8ParaNum;            /* 参数总数 */
}__attribute__ ((__packed__))MSG_0X000A_S;


typedef int (*HandleHook_Func)(unsigned char*, unsigned short,unsigned char);

typedef struct
{
    unsigned int u32MsgID;
    HandleHook_Func pFuncHandle;
}MSG_HANDLE_HOOK_S;


int IPCPMsg_Debug_Cmd1(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_Cmd2(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_Cmd3(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_Cmd4(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_Cmd5(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_Cmd6(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_Cmd7(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_Cmd8(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_Cmd9(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);
int IPCPMsg_Debug_CmdA(unsigned char *pu8SomeText, unsigned short u16Len, unsigned char u8SrcAddr);

#endif

测试：

    为了测试该方法的可靠和稳定性，有新建2个本地进程和2个远程设备进程，让他们随机时间随机往某个进程发送命令，看是否会出现命令消息丢失或是解析错误的问题。其中本地一个客户端的实现如下：

/************************************************************
*Copyright (C),lcb0281at163.com lcb0281atgmail.com
*FileName: local_client1.c
*BlogAddr: https://blog.csdn.net/li_wen01
*Description:本地模块进程1
*Date:	   2019-08-03
*Author:   Caibiao Lee
*Version:  V1.0
*Others:
*History:
***********************************************************/
#include <sys/types.h>
#include <sys/socket.h>
#include <stdio.h>
#include <sys/un.h>
#include <unistd.h>
#include <stdlib.h>
#include <signal.h>

#include "ipc_interface.h"
#include "ipc_msgstruct.h"
#include "ipc_common.h"

#define CLIENT_MODULE_ADDR		   LOCAL_CLIENT_1_ADDR
#define LOCAL1_DELAY_FOR_DEBUG_US  (1000*1000)

/**消息处理**/
MSG_HANDLE_HOOK_S g_astLocalMsgTable[] =
{
	{QUEUE_DEBUG_CMD1, IPCPMsg_Debug_Cmd1},
	{QUEUE_DEBUG_CMD2, IPCPMsg_Debug_Cmd2},
	{QUEUE_DEBUG_CMD3, IPCPMsg_Debug_Cmd3},
	{QUEUE_DEBUG_CMD4, IPCPMsg_Debug_Cmd4},
	{QUEUE_DEBUG_CMD5, IPCPMsg_Debug_Cmd5},
	{QUEUE_DEBUG_CMD6, IPCPMsg_Debug_Cmd6},
	{QUEUE_DEBUG_CMD7, IPCPMsg_Debug_Cmd7},
	{QUEUE_DEBUG_CMD8, IPCPMsg_Debug_Cmd8},
	{QUEUE_DEBUG_CMD9, IPCPMsg_Debug_Cmd9},
	{QUEUE_DEBUG_CMDA, IPCPMsg_Debug_CmdA},
    {0,NULL},
};

/**********************************
linux ctrl + C 会产生 SIGINT信号
接收到SIGINT 信号进入该函数
**********************************/
void stop(int signo)
{
	int i = 0;
	printf(" stop \n");

	_exit(0);
}

/*********************************************
当客户端断开连接的时候，
在服务端socket send进程可以收到收到信号SIGPIPE，
收到SIGPIPE信号进入该函数结束创建的线程。
**********************************************/
void signal_pipe(int signo)
{

}

/********************************************************
Function:	 ClientSendMsg
Description: 客户端(某一进程)发送数据包
Input:
	s32TargModuleAddr ： 需要发送到的模块地址
OutPut: none
Return: 0 成功;非0 异常
Others:
Author: Caibiao Lee
Date:	2019-08-03
*********************************************************/
int ClientSendMsg(int s32TargModuleAddr)
{
	int l_s32Res = 0;
	MSG_PACK_S l_stMsg = {0};
	MSG_0X0001_S l_stMsgSend = {0};

	if((s32TargModuleAddr<LOCAL_CLIENT_1_ADDR)&&(s32TargModuleAddr>LOCAL_CLIENT_MAX_ADDR))
	{
		printf("%s %d input para error %d \n",__FUNCTION__,__LINE__,s32TargModuleAddr);
		return -1;
	}

	l_stMsgSend.arrCardNo;
	l_stMsgSend.u16Altitude   = 0x66;
	l_stMsgSend.u16Direct     = 0x11;
	l_stMsgSend.u16SensorSpeed= 0x12;
	l_stMsgSend.u16SpeedX10   = 0x13;
	l_stMsgSend.u32Alarm      = 0xbb;
	l_stMsgSend.u32Latitude   = 0xaa;
	l_stMsgSend.u32Longtitude = 0x15;
	l_stMsgSend.u32Status     = 0x01;
	l_stMsgSend.u8CarCor	  = 0x08;

	l_stMsg.CmdId    = QUEUE_DEBUG_CMD1;
	l_stMsg.Data     = (unsigned char*)&l_stMsgSend;
	l_stMsg.Len      = sizeof(MSG_0X0001_S);
	l_stMsg.SrcAddr  = CLIENT_MODULE_ADDR;
	l_stMsg.TargAddr = s32TargModuleAddr;
	IPCP_Arch_Msg_PackSend(CLIENT_MODULE_ADDR,&l_stMsg);

	return 0;

}

/********************************************************
Function:	 ClientRecvMsg
Description: 客户端(某一进程)接收数据包，并对数据包进行解析
Input:
OutPut: none
Return: 0 成功;非0 异常
Others:
Author: Caibiao Lee
Date:	2019-08-03
*********************************************************/
int ClientRecvMsg(void)
{
	int i = 0;
	int l_s32RecvLen = 0;
	int l_s32SrcAddr = 0;
	int l_s32CmdId   = 0;
	int l_s32TextLen = 0;
	int l_s32FlowNum = 0;
	int l_u32IPCPMsgNum = 0;

	unsigned char *l_pu8MsgBody = NULL;
	ARCH_MSG_S l_stMsg = {0};

	l_s32RecvLen = IPCP_Arch_Msg_Recv(CLIENT_MODULE_ADDR,&l_stMsg);

	if(l_s32RecvLen>0)
	{
		IPCP_Arch_Msg_PlatformStartP(&l_pu8MsgBody, l_stMsg.SomeText);

		l_u32IPCPMsgNum = ARRAY_SIZE(g_astLocalMsgTable);
		l_s32SrcAddr = IPCP_Arch_Msg_AnalyzeGetSrcAddr(&l_stMsg);
		l_s32CmdId   = IPCP_Arch_Msg_AnalyzeGetCmdID(&l_stMsg);
		l_s32TextLen = IPCP_Arch_Msg_AnalyzeGetLen(&l_stMsg);
		l_s32FlowNum = IPCP_Arch_Msg_AnalyzeGetRecFlow(&l_stMsg);

		printf("l_s32SrcAddr = %d \n",l_s32SrcAddr);
		printf("l_s32CmdId   = %d \n",l_s32CmdId);
		printf("l_s32TextLen = %d \n",l_s32TextLen);
		printf("l_s32FlowNum = %d \n",l_s32FlowNum);

		for(i = 0; i < l_u32IPCPMsgNum; i++)
		{
			if(l_s32CmdId == g_astLocalMsgTable[i].u32MsgID)
			{
				if(NULL!=g_astLocalMsgTable[i].pFuncHandle)
				{
					g_astLocalMsgTable[i].pFuncHandle(l_pu8MsgBody, l_s32TextLen,l_s32SrcAddr);
				}
				break;
			}
		}
	}
	return 0;
}

int main(int argc,char *argv[])
{
	int l_s32Delay    = 0;
	int l_s32Addr     = 0;
	int l_as32SendAddr[3] ={0};

	l_as32SendAddr[0] = LOCAL_CLIENT_2_ADDR;
	l_as32SendAddr[1] = DEV1_CLIENT_1_ADDR,
	l_as32SendAddr[2] = DEV1_CLIENT_2_ADDR;

	/**注册 SIGPIPE信号**/
	signal(SIGPIPE,signal_pipe);

	/**注册SIGINT 信号**/
	signal(SIGINT,stop);

	while(1)
	{

		l_s32Addr = IPCP_GetRandomReal(0,3);
		if((l_s32Addr>=0)&&(l_s32Addr<=2))
		{
			ClientSendMsg(l_as32SendAddr[l_s32Addr]);
		}

		ClientRecvMsg();

		l_s32Delay = IPCP_GetRandomReal(50,100);
		usleep(l_s32Delay*1000);
	}
	return 0;
}

实际运行结果如下：

一种基于Unix Domain和TCP连接的跨设备多进程间通信的方法-LMLPHP

    上面测试是以50~100ms随机发送一包数据来测试，运行24小时后未检测到异常。

总结：

(1)阻塞与非阻塞问题

    实际客户端在接收和发送，应该是需要非阻塞的模式，正常的数据接收和发送不能影响客户端原本的业务逻辑处理。在非阻塞状态下数据的接收和发送都需要特别的注意。
    在发送的时候，因为是非阻塞的，当对方接收缓存满了的时候，发送端会收到EAGAIN的错误码，该错误码是让你进行数据的重新发送。在我这里我是设计发送失败会重新发送3遍，然后3次发送后还是失败，那么这包数据就会被丢弃。
    在接收的时候，我们需要去判断对方网络是否还处于连接的状态，可以使用select实现，也可以通过接收PIPE信号来判断，为了更好的数据处理，在我这里是使用select来判断对方网络是否已经断开连接。

(2)缓存大小问题

    每建立一个socket，在连接或是绑定之前，我们都可以设置该socket连接的收发缓存大小，可以根据实际应用的发送峰值来判断应该设置多大的缓存。在我Ubuntu16.04系统下，默认的TCP和UnixDomain收发缓存大小如下：

biao@ubuntu:~/test/ipcp_socket/server$ ./server
create unix domin socket lpthread success
create TCP socket lpthread success
create read write lpthread seccess
TCP SO_RCVBUF = 87380
TCP SO_SNDBUF = 16384
server waiting for tcp client connect
Unix Domain SO_RCVBUF = 212992
Unix Domain SO_SNDBUF = 212992
server waiting for unix domain client connect

以上，完整代码文件如下：

biao@ubuntu:~/test/ipcp_socket$ tree
.
├── common
│   ├── ipc_common.c
│   ├── ipc_common.h
│   ├── ipc_interface.c
│   ├── ipc_interface.h
│   ├── ipc_msgstruct.c
│   └── ipc_msgstruct.h
├── device1
│   ├── device1_client1.c
│   ├── device1_client2.c
│   └── Makefile
├── local
│   ├── local_client1.c
│   ├── local_client2.c
│   └── Makefile
├── server
│   ├── Makefile
│   └── server.c
└── tmp
    ├── client_socket1
    ├── client_socket2
    └── server_socket

5 directories, 17 files
biao@ubuntu:~/test/ipcp_socket$

代码查看和下载：

代码GitHub地址：https://github.com/licaibiao/IPC_Socket

CSDN下载地址：https://download.csdn.net/download/li_wen01/11558180

 

08-28 07:07
Powered by LMLPHP ©2024 li_wen01 0.041270