python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python socket库通信功能

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

作者：代码写不完了

本文主要给大家介绍了如何使用python的socket库实现通信功能,这里简单的给每个客户端增加一个不重复的uid,客户端之间可以根据这个uid选择进行广播通信,感兴趣的小伙伴快来看看吧

1.前言

这是一篇介绍使用python的socket库实现通信功能的文章.系统主要分为客户端和服务端。这里简单的给每个客户端增加一个不重复的uid，客户端之间可以根据这个uid选择进行广播通信，即与所有在线的客户端通信。也可以选择与单一客户端通信。

2.技术介绍

(1) socket:

Socket（套接字）是在应用层和传输层之间的一个抽象层，它提供了对TCP/IP协议族进行访问的接口。socket可以看成是一种特殊的文件描述符，与文件描述符类似，开发者可以对它进行读写操作，从而实现与远程主机的通信和数据传输。

在网络编程中，Socket通常被用于实现客户端和服务器之间的通信。当一个程序需要向另一个程序发送数据时，它会将数据包装在一个Socket对象中，然后通过网络传输给对方。接收端程序收到数据后，会再次将数据解包，存入自己的Socket缓存当中。

在TCP/IP协议中，Socket是通过IP地址、端口号和协议类型来标识的。其中，IP地址用来唯一标识一台主机，端口号则表示该主机上具体的网络应用程序，协议类型指明该Socket使用的具体传输协议，如TCP或UDP等。

Socket编程比较灵活，可以根据不同的需求进行不同的设置和配置，例如设置Socket的超时时间、缓冲区大小、数据包大小等。同时，Socket也支持多种编程语言，如Python、Java和C/C++等。因此，在实现网络应用程序时，Socket是非常常用和重要的组件之一。

(2) Python的socket库:

Python的socket库是一个用于网络编程的标准库，它提供了一套高级的网络接口，使得开发人员可以轻松地创建网络应用程序。

该库允许Python程序通过创建Socket对象来完成网络通信，每个Socket对象都与一个特定的协议族相关联，如AF_INET（使用IPv4协议）或AF_INET6（使用IPv6协议）。

socket库提供了一系列函数，可以用于创建、连接、监听和发送数据等各种网络操作。其中，一些常用的函数包括：

  • socket() : 创建一个Socket对象，指定协议族、sock类型、协议等参数；
  • bind() : 将Socket对象与本地地址进行绑定；
  • listen() : 开始监听指定端口上的连接请求；
  • accept() : 接受客户端连接请求，并返回新的Socket对象和客户端地址；
  • connect() : 连接到远端主机的指定端口；
  • recv()  和  send() : 接收和发送数据；
  • close() : 关闭Socket连接。

使用Python的socket库，可以方便地实现基于TCP或UDP协议的网络编程。同时，socket库也支持基于多线程和异步IO的网络编程，可以用于处理高并发的网络应用程序。

3.系统实现

(1) 服务端(server.py)

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()

(2) 客户端1(client1.py)

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()

(3) 客户端2(client2.py)

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()

4.系统功能演示

(1)启动服务端和客户端

注意事项：需要需要先启动服务端 

(2) 通信功能

client1输入clent2的ID，在client2里输入clent1的ID，然后就可以开始通信了！

5.总结

这里提供了一个简单的实现使用python的socket库实现网络通信功能。

以上就是基于Python的socket库实现通信功能的示例代码的详细内容，更多关于Python socket库通信功能的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:
阅读全文