我正在使用 UNIX 套接字进行进程间通信（实际上是线程间通信）的套接字回调。我们使用套接字的方式与 TCP/UDP 相同。
下面的代码是用 c/obj-c 编写的，使用 posix 线程。将它翻译成 Swift/NSThread 应该不难。
请注意，下面的程序作为服务器端工作，这意味着程序会在客户端连接的地方创建套接字。一旦客户端连接到套接字，系统会自动接受连接并分配另一个文件描述符来读取/写入。套接字回调反映了这两个阶段的操作。最初我们创建套接字，然后添加为运行循环源，以便系统可以在客户端尝试连接时回调回调。系统接受，然后分配并告诉回调一个文件描述符以与客户端读取/写入。然后我们从读/写 fd 创建另一个运行循环源并添加到运行循环。当 rx/tx 数据准备好时，将调用第二个回调。
主线：
主线程创建 UNIX 套接字和工作线程。套接字 fd 作为工作线程的参数传递。
#import <stdio.h>
#import <string.h>
#import <stdlib.h>
#import <unistd.h>
#import <pthread.h>
#import <sys/socket.h>
#import <sys/un.h>
#import <sys/stat.h>
#import <sys/types.h>
#import <UIKit/UIKit.h>
#import <Foundation/Foundation.h>
int setup(const char *ipcNode) {
int sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
if (sockfd == -1) {
return -1;
}
struct sockaddr_un sa = {0};
sa.sun_len = sizeof(sa);
sa.sun_family = AF_UNIX;
strcpy(sa.sun_path, ipcNode);
remove(sa.sun_path);
if (bind(sockfd, (struct sockaddr*)&sa, sizeof(struct sockaddr_un)) == -1) {
close(sockfd);
return -1;
}
// start up worker thread
pthread_attr_t at;
pthread_attr_init(&at);
pthread_attr_setdetachstate(&at, PTHREAD_CREATE_DETACHED);
pthread_t th;
pthread_create(&th, &at, workerThread, (void *)(long)(sockfd));
return 1;
}
工作线程：
该程序用作服务器。因此，它等待客户端连接（通过 connect()）。一旦连接成功，系统会自动调用accept()并分配读/写fd与客户端通信。这个 fd 被传递给接受回调例程 socketDataCallback()。然后我们使用读/写 fd 创建另一个回调 clientDataCallback()。
// worker thread
//
void *workerThread(void *tprm) {
int sockfd = (int)tprm;
int retval = listen(sockfd, 1); // mark as "server" side. here, accept only 1 connection request at a time
if (retval != 0) {
return NULL;
}
// create CFSocket and register it as data source.
CFSocketRef socket = CFSocketCreateWithNative(kCFAllocatorDefault, sockfd, kCFSocketAcceptCallBack, socketDataCallback, nil);
// don't close native fd on CFSocketInvalidate
CFSocketSetSocketFlags(socket, CFSocketGetSocketFlags(socket) & ~kCFSocketCloseOnInvalidate);
// create run loop source
CFRunLoopSourceRef socketRunLoop = CFSocketCreateRunLoopSource(kCFAllocatorDefault, socket, 0);
// add to run loop
CFRunLoopAddSource(CFRunLoopGetCurrent(), socketRunLoop, kCFRunLoopCommonModes);
CFRelease(socketRunLoop);
CFRelease(socket);
CFRunLoopRun();
// not return here untill run loop stops
close(sockfd);
return NULL;
}
// socket connection w/ client side. create another data source and add to run-loop
//
void socketDataCallback(CFSocketRef s, CFSocketCallBackType callbackType, CFDataRef address, const void *data, void *info) {
CFSocketContext socketContext;
memset(&socketContext, 0, sizeof(CFSocketContext));
int clientfd = *((int *)data); // get file descriptor (fd)
socketContext.info = (void *)((long)clientfd); // set fd at info of socketContext
// create CFSocket for tx/rx w/ connected client
CFSocketRef socket = CFSocketCreateWithNative(kCFAllocatorDefault, clientfd, kCFSocketReadCallBack | kCFSocketWriteCallBack, clientDataCallback, &socketContext);
CFSocketDisableCallBacks(socket, kCFSocketWriteCallBack);
CFRunLoopSourceRef socketRunLoop = CFSocketCreateRunLoopSource(kCFAllocatorDefault, socket, 0);
CFRunLoopAddSource(CFRunLoopGetCurrent(), socketRunLoop, kCFRunLoopCommonModes);
CFRelease(socket);
CFRelease(socketRunLoop);
}
// data to/from client
//
void clientDataCallback(CFSocketRef s, CFSocketCallBackType callbackType, CFDataRef address, const void *data, void *info) {
if (callbackType & kCFSocketWriteCallBack) {
// your own tx data prcess here
// txDataCallback(s, callbackType, address, data, info);
}
if (!(callbackType & kCFSocketReadCallBack)) return;
// extract fd
int fd = (int)((long)info);
// read data, and do some work
uint8_t rxdata[1024];
size_t nr = read(fd, rxdata, 1024);
if (!nr) {
// socket closed
handleSocketClosed(s);
return;
}
// your own rx process here
}
// socket closed
//
void handleSocketClosed(CFSocketRef s) {
// any clean up process here, then
CFSocketInvalidate(s);
// stop run loop if necessary
// CFRunLoopStop(CFRunLoopGetCurrent());
}
如果您在客户端工作，事情会变得容易一些。您会通过 connect() 调用获得读/写 fd。然后你创建 CFSockeRef 并使用 fd 添加到运行循环。
希望这可以帮助。
编辑：如何等待 POSIX select()。在工作线程中使用 POSIX select() 等待比套接字回调更简单。如果您在客户端，则：
int sockfd = socket(...);
bind(sockfd, ...)
connect(sockfd, ...);
while (1) {
int nfds = sockfd+1;
fd_set rfds;
FD_ZERO(&rfds);
FD_SET(sockfd, &rfds);
int retval = select(nfds, &rfds, NULL, NULL, NULL);
if (retval == -1) break;
if (retval > 0) {
uint8_t rxdata[1024];
size_t nr = read(sockfd, rxdata, 1024);
if (!nr) {
// socket closed.
break;
}
// do your rx process here
}
}
在您的工作线程上运行上面的代码。