基于Python的socket库实现通信功能的示例代码

import socket
import threading
import json
HOST = '127.0.0.1'
PORT = 8888
# 存储客户端的 uid 和 socket 对象的映射关系
clients = {}
# 创建一个 TCP 协议的套接字对象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定并监听一个指定的地址和端口号
server_socket.bind((HOST, PORT))
# 开始监听客户端对指定 IP 地址和端口号的连接请求,当有客户端发送连接请求时，
# 服务端可以接受客户端连接，并创建一个新的套接字对象以处理客户端与服务端之间的通信。
server_socket.listen()
def handle_client(client_socket, client_address):
    uid = client_socket.recv(1024).decode('utf-8')
    print('uid:{}已上线'.format(uid))
    clients[uid] = client_socket
    while True:
        try:
            data = client_socket.recv(1024)
            if data:
                # 解析消息
                msg_dict = json.loads(data.decode('utf-8'))
                sender_id = msg_dict['sender_id']
                receiver_id = msg_dict['receiver_id']
                message = msg_dict['message']
                print('用户 {} 发送给用户 {} 的消息: {}'.format(sender_id, receiver_id, message))
                if receiver_id == 'all':
                    # 广播消息给所有客户端
                    for uid, client in clients.items():
                        if uid != sender_id:
                            client.sendall(data)
                elif receiver_id in clients and receiver_id != sender_id:
                    # 发送消息给指定客户端
                    client_socket = clients[receiver_id]
                    client_socket.sendall(data)
                else:
                    # 接收到的消息有误，回复错误信息
                    error = {
                        '错误': f'对方ID无效: {receiver_id}'
                    }
                    client_socket.sendall(json.dumps(error).encode('utf-8'))
            else:
                # 客户端关闭了连接
                print('客户端关闭了连接')
                break
        except:
            # 发送和接收消息时出现异常，可能是客户端关闭了连接
            print('发送和接收消息时出现异常，可能是客户端关闭了连接')
            break
    # 移除客户端连接
    print('用户:{}已离线'.format(uid))
    del clients[uid]
    # 关闭客户端连接
    client_socket.close()
if __name__ == '__main__':
    print(f'Server listening on {HOST}:{PORT}')
    while True:
        client_socket, client_address = server_socket.accept()
        print(f'{client_address[0]}:{client_address[1]} connected')
        # 生成客户端 UID，并存储客户端的 socket 对象和 UID 的映射关系
        thread = threading.Thread(target=handle_client, args=(client_socket, client_address))
        thread.daemon = True
        thread.start()

import socket
import threading
import uuid
import json
HOST = '127.0.0.1'
PORT = 8888
# 随机生成一个唯一 ID
UID = str(uuid.uuid4())
# 创建一个 TCP 协议的套接字对象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 连接指定主机和端口
client_socket.connect((HOST, PORT))
# 将UID发送给服务端
client_socket.sendall(UID.encode('utf-8'))
def receive_message():
    while True:
        data = client_socket.recv(1024)
        if data:
            # 解析消息
            msg_dict = json.loads(data.decode('utf-8'))
            if 'sender_id' in msg_dict:
                sender_id = msg_dict['sender_id']
                message = msg_dict['message']
                print(f"{sender_id}: {message}")
            elif 'error' in msg_dict:
                error = msg_dict['error']
                print(f"Error: {error}")
        else:
            # 服务端关闭了连接
            print('服务器已下线...')
            break
    # 关闭客户端连接
    client_socket.close()
def send_message(receiver_id):
    while True:
        message = input()
        # 构造消息字典
        msg_dict = {
            'sender_id': UID,
            'receiver_id': receiver_id,
            'message': message
        }
        # 将消息字典序列化为JSON格式字符串
        json_str = json.dumps(msg_dict)
        # 将消息发送给服务端
        client_socket.sendall(json_str.encode('utf-8'))
if __name__ == '__main__':
    print(f'你的ID: {UID}')
    receiver_id = input('输入对方的ID接收对方消息或者输入"all"接收广播消息: ')
    # 启动接收消息的线程
    t1 = threading.Thread(target=receive_message)
    t1.daemon = True
    t1.start()
    # 启动发送消息的线程
    t2 = threading.Thread(target=send_message, args=(receiver_id,))
    t2.daemon = True
    t2.start()
    # 等待发送线程结束，结束接收线程
    t2.join()
    t1.join()

import socket
import threading
import uuid
import json
HOST = '127.0.0.1'
PORT = 8888
# 随机生成一个唯一 ID
UID = str(uuid.uuid4())
# 创建一个 TCP 协议的套接字对象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 连接指定主机和端口
client_socket.connect((HOST, PORT))
# 将UID发送给服务端
client_socket.sendall(UID.encode('utf-8'))
def receive_message():
    while True:
        data = client_socket.recv(1024)
        if data:
            # 解析消息
            msg_dict = json.loads(data.decode('utf-8'))
            if 'sender_id' in msg_dict:
                sender_id = msg_dict['sender_id']
                message = msg_dict['message']
                print(f"{sender_id}: {message}")
            elif 'error' in msg_dict:
                error = msg_dict['error']
                print(f"Error: {error}")
        else:
            # 服务端关闭了连接
            print('服务器已下线...')
            break
    # 关闭客户端连接
    client_socket.close()
def send_message(receiver_id):
    while True:
        message = input()
        # 构造消息字典
        msg_dict = {
            'sender_id': UID,
            'receiver_id': receiver_id,
            'message': message
        }
        # 将消息字典序列化为JSON格式字符串
        json_str = json.dumps(msg_dict)
        # 将消息发送给服务端
        client_socket.sendall(json_str.encode('utf-8'))
if __name__ == '__main__':
    print(f'你的ID: {UID}')
    receiver_id = input('输入对方的ID接收对方消息或者输入"all"接收广播消息: ')
    # 启动接收消息的线程
    t1 = threading.Thread(target=receive_message)
    t1.daemon = True
    t1.start()
    # 启动发送消息的线程
    t2 = threading.Thread(target=send_message, args=(receiver_id,))
    t2.daemon = True
    t2.start()
    # 等待发送线程结束，结束接收线程
    t2.join()
    t1.join()