代码收藏家技术教程 2024-12-28

【Python】多人聊天室案例、生成器和迭代器

多人聊天室案例

需求

服务器只有一个，打开界面后点击“启动”，开启socket服务

每个人都可以打开一个客户端界面，每个人进来之前都必须定义一个名字

客户端点击“连接”，进入聊天室，进入之后，服务器发出通知，每个人都能看到

客户端可以发送一条消息到聊天室，所有人都可以看到刚刚发送的该条消息

客户端可以点击“离开”退出聊天室。服务器也要发出通知

服务器点击“聊天记录保存”，保存聊天信息到文件中

服务器点击“停止”，所有人都退出聊天室（发通知），并且服务器窗口关闭

实现

我们可以使用Python的socket库来处理网络通信，并结合PyQt5来创建图形用户界面（GUI）。我们将构建一个服务器端和多个客户端，每个客户端都可以连接到服务器进行消息发送和接收。以下是具体的实现步骤：

服务端代码（server.py）

Server 类：负责管理所有客户端连接、接收并广播消息以及处理客户端加入和离开的通知。

MainWindow 类：提供图形用户界面，包括启动/停止服务器按钮、显示聊天信息的文本框等。

import socket
import threading
from PyQt5.QtWidgets import QApplication, QMainWindow, QTextEdit, QPushButton, QVBoxLayout, QWidget, QLineEdit, QLabel
from PyQt5.QtCore import pyqtSignal, QObject

class Server(QObject):
    new_message = pyqtSignal(str)
    client_connected = pyqtSignal(str)
    client_disconnected = pyqtSignal(str)

    def __init__(self):
        super().__init__()
        self.clients = {}
        self.server_socket = None
        self.running = False

    def start_server(self):
        if not self.running:
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.bind(('localhost', 12345))
            self.server_socket.listen(5)
            self.running = True
            threading.Thread(target=self.accept_clients).start()
            self.new_message.emit("Server started.")

    def accept_clients(self):
        while self.running:
            client_socket, addr = self.server_socket.accept()
            client_name = client_socket.recv(1024).decode('utf-8')
            self.clients[client_name] = client_socket
            self.client_connected.emit(f"{client_name} joined the chat.")
            threading.Thread(target=self.handle_client, args=(client_name,)).start()

    def handle_client(self, client_name):
        client_socket = self.clients[client_name]
        while self.running:
            try:
                message = client_socket.recv(1024).decode('utf-8')
                if message.lower() == "leave":
                    break
                self.broadcast_message(f"{client_name}: {message}")
            except ConnectionResetError:
                break
        self.remove_client(client_name)

    def broadcast_message(self, message):
        for name, sock in self.clients.items():
            try:
                sock.sendall(message.encode('utf-8'))
            except:
                pass
        self.new_message.emit(message)

    def remove_client(self, client_name):
        if client_name in self.clients:
            del self.clients[client_name]
            self.client_disconnected.emit(f"{client_name} left the chat.")
            self.broadcast_message(f"{client_name} has left the chat.")

    def stop_server(self):
        self.running = False
        for client_socket in self.clients.values():
            client_socket.close()
        self.server_socket.close()
        self.new_message.emit("Server stopped.")

    def save_chat_log(self, chat_log):
        with open("chat_log.txt", "w") as f:
            f.write(chat_log)
        self.new_message.emit("Chat log saved.")

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Chat Server")
        self.setGeometry(100, 100, 600, 400)

        self.server = Server()
        self.server.new_message.connect(self.update_chat)
        self.server.client_connected.connect(self.update_chat)
        self.server.client_disconnected.connect(self.update_chat)

        self.chat_display = QTextEdit(self)
        self.chat_display.setReadOnly(True)

        self.start_button = QPushButton("启动", self)
        self.start_button.clicked.connect(self.server.start_server)

        self.stop_button = QPushButton("停止", self)
        self.stop_button.clicked.connect(self.stop_and_close)

        self.save_log_button = QPushButton("聊天记录保存", self)
        self.save_log_button.clicked.connect(self.save_chat_log)

        layout = QVBoxLayout()

客户端代码（client.py）

Client 类：用于与服务器建立连接，接收来自服务器的消息，并向服务器发送消息。

MainWindow 类：提供图形用户界面，允许用户输入名字、连接/断开服务器、发送消息及查看聊天内容。

import socket
import threading
from PyQt5.QtWidgets import QApplication, QMainWindow, QTextEdit, QPushButton, QVBoxLayout, QWidget, QLineEdit, QLabel, QMessageBox
from PyQt5.QtCore import pyqtSignal, QObject

class Client(QObject):
    new_message = pyqtSignal(str)

    def __init__(self, name):
        super().__init__()
        self.name = name
        self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.running = False

    def connect_to_server(self):
        try:
            self.client_socket.connect(('localhost', 12345))
            self.client_socket.sendall(self.name.encode('utf-8'))
            self.running = True
            threading.Thread(target=self.receive_messages).start()
            return True
        except:
            return False

    def receive_messages(self):
        while self.running:
            try:
                message = self.client_socket.recv(1024).decode('utf-8')
                self.new_message.emit(message)
            except:
                break

    def send_message(self, message):
        if self.running:
            self.client_socket.sendall(message.encode('utf-8'))

    def disconnect(self):
        self.running = False
        self.client_socket.sendall("leave".encode('utf-8'))
        self.client_socket.close()

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Chat Client")
        self.setGeometry(100, 100, 600, 400)

        self.name_input = QLineEdit(self)
        self.name_input.setPlaceholderText("Enter your name")

        self.connect_button = QPushButton("连接", self)
        self.connect_button.clicked.connect(self.connect_to_chat)

        self.disconnect_button = QPushButton("离开", self)
        self.disconnect_button.clicked.connect(self.disconnect_from_chat)
        self.disconnect_button.setEnabled(False)

        self.message_input = QLineEdit(self)
        self.message_input.setPlaceholderText("Type a message")

        self.send_button = QPushButton("发送", self)
        self.send_button.clicked.connect(self.send_message)
        self.send_button.setEnabled(False)

        self.chat_display = QTextEdit(self)
        self.chat_display.setReadOnly(True)

        layout = QVBoxLayout()
        layout.addWidget(QLabel("Name:"))
        layout.addWidget(self.name_input)
        layout.addWidget(self.connect_button)
        layout.addWidget(self.disconnect_button)
        layout.addWidget(self.message_input)
        layout.addWidget(self.send_button)
        layout.addWidget(self.chat_display)

        container = QWidget()
        container.setLayout(layout)
        self.setCentralWidget(container)

        self.client = None

    def connect_to_chat(self):
        name = self.name_input.text().strip()
        if name:
            self.client = Client(name)
            if self.client.connect_to_server():
                self.client.new_message.connect(self.update_chat)
                self.name_input.setEnabled(False)
                self.connect_button.setEnabled(False)
                self.disconnect_button.setEnabled(True)
                self.message_input.setEnabled(True)
                self.send_button.setEnabled(True)
                self.update_chat(f"Connected as {name}.")
            else:
                QMessageBox.critical(self, "Error", "Could not connect to server.")
        else:
            QMessageBox.warning(self, "Warning", "Please enter a name.")

    def disconnect_from_chat(self):
        if self.client:
            self.client.disconnect()
            self.client = None
            self.name_input.setEnabled(True)
            self.connect_button.setEnabled(True)
            self.disconnect_button.setEnabled(False)
            self.message_input.setEnabled(False)
            self.send_button.setEnabled(False)
            self.update_chat("Disconnected.")

    def send_message(self):
        message = self.message_input.text().strip()
        if message:
            self.client.send_message(message)
            self.message_input.clear()

    def update_chat(self, message):
        self.chat_display.append(message)

if __name__ == "__main__":
    app = QApplication([])
    window = MainWindow()
    window.show()
    app.exec_()

通过这种方式，你可以运行一个服务器实例和多个客户端实例，实现多人在线聊天的功能。当服务器点击“聊天记录保存”时，它会将当前聊天记录保存到文件中；而当服务器点击“停止”时，所有客户端都会收到通知并自动断开连接。

第十七节：生成器和迭代器

在 Python 中，生成器（Generators）和迭代器（Iterators）是用于逐个访问元素序列的工具。它们使得你可以编写高效且易于维护的代码来处理大量数据或无限流的数据，而无需一次性将所有数据加载到内存中。下面详细介绍这两种概念及其区别。

一、迭代器

定义：

迭代器是一个实现了 iter() 和 next() 方法的对象。

iter() 返回迭代器对象本身。

next() 返回序列中的下一个项目；如果没有更多项，则抛出 StopIteration 异常。

用途：

迭代器允许你遍历一个容器（如列表、元组、字典等），但是一次只获取一个元素。

它们提供了一种抽象的方式，让开发者可以轻松地遍历复杂的数据结构。

创建方式

可以通过实现上述两个特殊方法来自定义迭代器类。

或者使用内置函数如 iter() 来从可迭代对象（例如列表、字符串）创建迭代器。

本质： 一个实现了__iter__方法和__next__方法的对象

注意 Iterator对象和 Iterable对象，一个是迭代器，一个是可迭代对象

1、list、dict、str、tuple、set是可迭代对象但不是迭代器；

2、可迭代对象可以转为迭代器，for循环会自动转换成迭代器。或者调用iter函数

3、如果把所有数据丢到列表中可以优点速度快缺点列表占内存太大,如果使用迭代器申请固定的空间也就是一个个的拿出来，能节约内存，但是浪费时间；

4、需要用类来写迭代器，需要重写 _ iter _( ) 和 _ next _( ）方法；思考一下

5、自定义迭代器最大的特点是，需要用类来写，显得代码冗长，不太方便。所以直接使用生成器

from collections.abc import Iterable, Iterator
class GenreratorPrime(object):

    def __init__(self):
        self.i = 2

	# 需要用类来写迭代器，需要重写 _ _iter_ _( ) 和 _ _next_ _( ）方法；
    def __iter__(self):
        return self

    def __next__(self):
        if self.i == 2:
            self.i += 1
            return 2

        while True:
            self.i += 1
            for j in range(2, self.i):
                if self.i % j == 0:
                    break
            else:
                return self.i

gp = GenreratorPrime()
print(isinstance(gp, Iterable))
print(isinstance(gp, Iterator))
iter1 = iter(gp)
print(next(iter1))
print(next(iter1))
print(next(iter1))
print(next(iter1))
print(next(iter1))
print(next(iter1))
print(next(iter1))
print(next(iter1))

二、生成器

定义：

生成器是一种特殊的迭代器，它由包含 yield 表达式的函数定义。

当调用生成器函数时，它不会立即执行函数体内的代码，而是返回一个生成器对象。

每次调用生成器的 next() 方法时，函数会从上次暂停的地方继续执行，直到遇到下一个 yield 语句为止。

优点：

简化了迭代器的创建过程，因为不需要显式地实现 iter() 和 next() 方法。

更节省内存，因为它生成值是在需要的时候才计算出来，而不是预先全部生成并存储起来。

支持惰性求值（lazy evaluation），这对于处理大数据集或者无限流的数据非常有用。

创建方式

使用带有 yield 关键字的函数来定义生成器。

列表推导式的语法也可以用来创建生成器表达式（只需将方括号替换为圆括号）。

生成器（generator）也是一种迭代器，在每次迭代时返回一个值，直到抛出 StopIteration 异常。它有两种构造方式：

表达式来创建生成器

包含有yield的函数来创建生成器

1、表达式创建生成器

print([x for x in range(6)])
print((x for x in range(6)))
numbers = (x for x in range(6))
# for n in numbers:
#     print(n)

print(hasattr(numbers, '__iter__'))
print(hasattr(numbers, 'next'))
print(hasattr(numbers, '__next__'))

print(numbers.__next__())

print(next(numbers))
print(next(numbers))
print(next(numbers))
print(next(numbers))
print(next(numbers))
print(next(numbers))
print(next(numbers))

总结：

可以看出生成器表达式无法像列表推导式那样直接输出，它和可迭代对象一样只能采用for循环调用next（）函数，原因在于range返回的是一个可迭代对象，列表推导式之所以能直接print就是因为[]将可迭代对象转为列表。

2、含有yield关键字的函数

一个带有 yield 的函数就是一个生成器函数，当我们使用 yield 时，它帮我们自动创建了__iter__() 和 next() 方法，而且在没有数据时，也会抛出 StopIteration 异常，也就是我们不费吹灰之力就获得了一个迭代器，非常简洁和高效。

def generator_func():
    v1 = yield 1
    print(f'hello {v1}')
    v1 = yield 2
    print(f'value1 is {v1}')
    v2 = yield 3
    print(f'value2 is {v2}')
    v3 = yield 4
    print(f'value3 is {v3}')


g = generator_func()
print(g.__next__())
print(g.__next__())
print(g.send(100))
g.send(1)
print(g.send(2))

总结：

yield 把函数变成了一个生成器。
调用该函数的时候不会立即执行代码，而是返回了一个生成器对象
当使用 next() (在 for 循环中会自动调用 next() ) 作用于返回的生成器对象时，函数开始执行，在遇到 yield 的时候会『暂停』，并返回当前的迭代值；
当再次使用 next() 的时候，函数会从原来『暂停』的地方继续执行，直到遇到 yield语句，如果没有 yield 语句，则抛出异常；
生成器函数的执行过程看起来就是不断地 =执行->中断->执行->中断的过程
send() 方法就是 next() 的功能，加上传值给上次暂停的yield 。
close() 方法来关闭一个生成器。生成器被关闭后，再次调用 next() 方法，不管能否遇到 yield 关键字，都会抛出 StopIteration 异常，

3、案例

创建一个获取所有质数的生成器

def generator_prime():
    i = 2
    yield i
    while True:
        i += 1
        for j in range(2, i):
            if i % j == 0:
                break
        else:
            yield i


g = generator_prime()
print(next(g))
print(next(g))
print(next(g))
print(next(g))
print(next(g))
print(next(g))
print(next(g))

作者：道友老李