unix dgram通信

本文介绍: 【代码】unix dg ram 通信。

# 同一个机器多个进程间通信，unix比AF_INET效率更高
# client
import socket               # 导入 socket 模块
import os
import threading
import time

class SocketClient(threading.Thread):
    def __init__(self):
        super(SocketClient, self).__init__()
        
        self._addr = "/tmp/socket/unix_socket"
        self._socket = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_DGRAM)         # 创建 socket 对象

    def connect(self):
        self._socket.connect(self._addr)

    def run(self):
        while True:
            msg = "hello, I am is client socket"
            self._socket.sendall(msg.encode())
            # msg = self._socket.recv(1024)
            # print(f"client recv msg:{msg.decode()}")
            time.sleep(1)

if __name__ == "__main__":
    try:
        client = SocketClient()                
        client.connect()
    except Exception as e:
        print(f"exception {e}")
    else:
        client.start()

# server
import socket               # 导入 socket 模块
import os
import threading
import time

class SocketServer(threading.Thread):
    def __init__(self):
        super(SocketServer, self).__init__()
        
        # 使用socket文件进行单项通信， 数据报类型
        self._addr = "/tmp/socket/unix_socket"
        self._socket = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_DGRAM)         # 创建 socket 对象


    def bind(self):
        if os.path.exists(self._addr):
            os.unlink(self._addr)
        self._socket.bind(self._addr)

    def run(self):
        while True:
            msg = self._socket.recv(1024)
            print(f"server recv msg:{msg.decode()}")
            # self._socket.sendall(msg.encode())
            time.sleep(1)

if __name__ == "__main__":
    try:
        server = SocketServer()                
        server.bind()
    except Exception as e:
        print(f"exception {e}")
    else:
        server.start()