DBUS socket 进程间交互原理

基于 ​​UNIX Domain Socket​​ 的双向通信框架，核心原理模型如下：

Server[服务端] -->|监听| Socket[/tmp/dbus-name]
Client[客户端] -->|连接| Socket
Server --> Thread1[监听线程]
Client --> Thread2[方法线程]

服务端​​：创建监听套接字（socket+bind+listen），接受其他进程的连接
​​客户端​​：主动连接服务端（socket+connect），建立双向通道
​​多线程​​：使用独立线程处理连接监听（listenThread）和方法调用（methodThread）

1. 该函数实现基于 UNIX Domain Socket 的双向通信框架，主要完成：

• ​​服务端​​：创建监听套接字、绑定地址、启动监听线程
• ​​客户端​​：创建连接套接字、尝试重连、启动方法调用线程
• ​​线程安全​​：通过互斥锁 (dbus_mutex) 保护共享资源

• UNIX Domain Socket特点​​

​​ 高效性​​：内核直接传递数据，无需网络协议栈

​​ 可靠性​​：面向连接（SOCK_STREAM），保证数据顺序

​​ 安全性​​：通过文件系统权限控制访问（/tmp/dbus-*）

2.程序实现

#include <sys/un.h>
#include <fcntl.h>
#include <poll.h>

#define UNIX_PATH_MAX 108
#define MAX_RETRY_ATTEMPTS 100  // 50ms * 100 = 5s
#define DETACHED_THREAD 1

/* 服务端初始化 */

static int init_server(ZRTDbusInfo dbusInfo) {
struct sockaddr_un srv_addr;
pthread_attr_t thread_attr;
// 创建UNIX域套接字
dbusInfo->socket_server = socket(AF_UNIX, SOCK_STREAM, 0);
if (dbusInfo->socket_server < 0) {
    perror("socket() server failed");
    return -1;
}

// 生成安全的套接字路径
if (snprintf(dbusInfo->unix_domain, UNIX_PATH_MAX, 
            "/tmp/dbus-%.32s", dbusInfo->name) >= UNIX_PATH_MAX) {
    fprintf(stderr, "Unix domain path too long\n");
    close(dbusInfo->socket_server);
    return -1;
}

// 清理旧套接字文件
unlink(dbusInfo->unix_domain);

// 配置地址结构
memset(&srv_addr, 0, sizeof(srv_addr));
srv_addr.sun_family = AF_UNIX;
strncpy(srv_addr.sun_path, dbusInfo->unix_domain, sizeof(srv_addr.sun_path)-1);
srv_addr.sun_path[sizeof(srv_addr.sun_path)-1] = '\0';

// 绑定地址
if (bind(dbusInfo->socket_server, (struct sockaddr*)&srv_addr, 
        sizeof(srv_addr)) < 0) {
    perror("bind() failed");
    close(dbusInfo->socket_server);
    return -1;
}

// 设置监听队列
if (listen(dbusInfo->socket_server, MAX_LISTEN_NUM) < 0) {
    perror("listen() failed");
    close(dbusInfo->socket_server);
    unlink(dbusInfo->unix_domain);
    return -1;
}

// 创建分离线程
pthread_attr_init(&thread_attr);
pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED);
if (pthread_create(&dbusInfo->listenThread_t, &thread_attr, 
                  ZRT_Dbus_Listen_Thread, dbusInfo) != 0) {
    pthread_attr_destroy(&thread_attr);
    close(dbusInfo->socket_server);
    unlink(dbusInfo->unix_domain);
    return -1;
}
pthread_attr_destroy(&thread_attr);

return 0;

}

/* 客户端初始化（带非阻塞连接） */

static int init_client(ZRTDbusInfo dbusInfo) {
struct sockaddr_un srv_addr;
char server_path[UNIX_PATH_MAX];
pthread_attr_t thread_attr;
int flags, ret;
// 创建客户端套接字
if ((dbusInfo->socket_client = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
    perror("socket() client failed");
    return -1;
}

// 生成服务端路径
snprintf(server_path, UNIX_PATH_MAX, "/tmp/dbus-%.32s", dbusInfo->dst_name);

// 配置地址结构
memset(&srv_addr, 0, sizeof(srv_addr));
srv_addr.sun_family = AF_UNIX;
strncpy(srv_addr.sun_path, server_path, sizeof(srv_addr.sun_path)-1);
srv_addr.sun_path[sizeof(srv_addr.sun_path)-1] = '\0';

// 设置为非阻塞模式
flags = fcntl(dbusInfo->socket_client, F_GETFL, 0);
fcntl(dbusInfo->socket_client, F_SETFL, flags | O_NONBLOCK);

// 带超时的连接
int connected = 0;
struct pollfd pfd = { .fd = dbusInfo->socket_client, .events = POLLOUT };

for (int i = 0; i < MAX_RETRY_ATTEMPTS; ++i) {
    ret = connect(dbusInfo->socket_client, (struct sockaddr*)&srv_addr, 
                 sizeof(srv_addr));
    if (ret == 0) {
        connected = 1;
        break;
    }

    if (errno == EINPROGRESS) {
        // 使用poll等待连接完成
        if (poll(&pfd, 1, 50) > 0 && (pfd.revents & POLLOUT)) {
            socklen_t len = sizeof(ret);
            getsockopt(dbusInfo->socket_client, SOL_SOCKET, SO_ERROR, &ret, &len);
            if (ret == 0) {
                connected = 1;
                break;
            }
        }
    }
    usleep(50000); // 50ms
}

if (!connected) {
    fprintf(stderr, "Connection timeout to %s\n", dbusInfo->dst_name);
    close(dbusInfo->socket_client);
    dbusInfo->socket_client = -1;
    return -1;
}

// 恢复阻塞模式
fcntl(dbusInfo->socket_client, F_SETFL, flags);

// 创建方法线程
pthread_attr_init(&thread_attr);
pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED);
if (pthread_create(&dbusInfo->methodThread_t, &thread_attr, 
                  ZRT_Dbus_Method_Thread, dbusInfo) != 0) {
    pthread_attr_destroy(&thread_attr);
    close(dbusInfo->socket_client);
    dbusInfo->socket_client = -1;
    return -1;
}
pthread_attr_destroy(&thread_attr);

return 0;
}

/* 主注册函数 */

int ZRT_Dbus_Register(ZRTDbusInfo *dbusInfo) {
// 参数校验
if (!dbusInfo || !dbusInfo->name[0]) {
fprintf(stderr, "Invalid parameters\n");
return -1;
}
// 初始化互斥锁
if (pthread_mutex_init(&dbusInfo->dbus_mutex, NULL) != 0) {
    perror("pthread_mutex_init failed");
    return -1;
}

// 服务端初始化
if (dbusInfo->interface_num > 0) {
    if (init_server(dbusInfo) < 0) {
        goto cleanup_mutex;
    }
}

// 客户端初始化
if (dbusInfo->dst_name[0]) {
    if (init_client(dbusInfo) < 0) {
        goto cleanup_server;
    }
}

return 0;

cleanup_server:
if (dbusInfo->socket_server >= 0) {
  close(dbusInfo->socket_server);
  unlink(dbusInfo->unix_domain);
}
cleanup_mutex:
  pthread_mutex_destroy(&dbusInfo->dbus_mutex);
  return -1;
}

3。 dbus 通信优势