Python网络编程（一）

Tcp/Ip编程

服务端
bind(Union) 绑定ip和端口
listen() 启动监听，监听来自客户端的连接
accept() 接收客户端sock
send() 发送信息
recv() 接收信息，阻塞
close() 关闭连接

客户端
connect() 连接服务端

1.0 server
from socket import socket, AF_INET, SOCK_STREAM
import struct
import json
from threading import Thread


class ServerTool(socket):
    socks = []

    def accept_sock(self):
        while True:
            sock, addr = self.accept()
            print('accept!')
            ServerTool.socks.append(sock)

    def __init__(self, family, type):
        super().__init__(family, type)
        self.bind(('127.0.0.1', 8888))
        self.listen(5)
        print("Server started...")
        Thread(target=self.accept_sock).start()

    def send_msg(self, msg):
        msg_bytes = bytes(msg, encoding='utf8')
        msg_dict = {
            'size': len(msg_bytes)
        }
        msg_json_str = json.dumps(msg_dict)
        msg_json_str_bytes = bytes(msg_json_str, encoding='utf8')
        msg_json_str_bytes_len = struct.pack('i', len(msg_json_str_bytes))
        for s in ServerTool.socks:
            s.send(msg_json_str_bytes_len)
            s.send(msg_json_str_bytes)
            s.send(msg_bytes)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def reader(self):
        while True:
            self.read_msg()

    def read_msg(self):
        msg_json_str_bytes_len = self.recv(4)
        if len(msg_json_str_bytes_len) < 4:
            return None
        msg_json_str_bytes_len = struct.unpack('i', msg_json_str_bytes_len)
        msg_json_str_bytes = self.recv(msg_json_str_bytes_len[0])
        msg_json_str = json.loads(msg_json_str_bytes.decode('utf8'))
        msg_bytes = self.recv(msg_json_str['size'])
        msg = msg_bytes.decode('utf8')
        print('receive:', msg)


with ServerTool(AF_INET, SOCK_STREAM) as server:
    msg = None
    while True:
        msg = input('输入群发内容...')
        if msg == 'exit':
            break
        server.send_msg(msg)

1.0 client
from socket import socket, AF_INET, SOCK_STREAM, _Address
import struct
import json
from typing import Union
from threading import Thread


class ClientTool(socket):

    def __init__(self, family, type):
        super().__init__(family, type)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def reader(self):
        while True:
            self.read_msg()

    def read_msg(self):
        msg_json_str_bytes_len = self.recv(4)
        if len(msg_json_str_bytes_len) < 4:
            return None
        msg_json_str_bytes_len = struct.unpack('i', msg_json_str_bytes_len)
        msg_json_str_bytes = self.recv(msg_json_str_bytes_len[0])
        msg_json_str = json.loads(msg_json_str_bytes.decode('utf8'))
        msg_bytes = self.recv(msg_json_str['size'])
        msg = msg_bytes.decode('utf8')
        print('receive:', msg)

    def connect(self, address: Union[_Address, bytes]) -> None:
        super().connect(address)
        Thread(target=self.reader).start()


with ClientTool(AF_INET, SOCK_STREAM) as client:
    client.connect(('127.0.0.1', 8888))
    while True:
        msg = input('输入群发内容...')
        if msg == 'exit':
            break
        client.send_msg(msg)

Python网络编程（一）

原文：https://www.cnblogs.com/hamusuta/p/12423696.html