MQ大牛成长课–从0到1手写分布式消息队列中间件（高の青）

import socket
import threading
import json
from queue import Queue

class MessageQueueServer:
    def __init__(self, host='localhost', port=6789):
        self.host = host
        self.port = port
        self.clients = []
        self.message_queue = Queue()
        self.running = False

    def listen_for_clients(self):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
            server_socket.bind((self.host, self.port))
            server_socket.listen()
            print(f"Server started on {self.host}:{self.port}")
            self.running = True
            while self.running:
                client_socket, addr = server_socket.accept()
                print(f"Connected by {addr}")
                self.clients.append(client_socket)
                threading.Thread(target=self.handle_client, args=(client_socket,)).start()

    def handle_client(self, client_socket):
        while self.running:
            try:
                message = client_socket.recv(1024).decode('utf-8')
                if message:
                    self.message_queue.put(message)
                    print(f"Received message: {message}")
                    self.broadcast_message(message)
            except:
                break

        client_socket.close()

    def broadcast_message(self, message):
        for client in self.clients:
            try:
                client.sendall(message.encode('utf-8'))
            except:
                self.clients.remove(client)

    def start_server(self):
        threading.Thread(target=self.listen_for_clients).start()

    def stop_server(self):
        self.running = False
        for client in self.clients:
            client.close()

if __name__ == "__main__":
    server = MessageQueueServer()
    server.start_server()

import socket
import threading

class MessageQueueClient:
    def __init__(self, host='localhost', port=6789):
        self.host = host
        self.port = port
        self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    def connect(self):
        self.client_socket.connect((self.host, self.port))
        print(f"Connected to server at {self.host}:{self.port}")

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

    def receive_messages(self):
        while True:
            try:
                message = self.client_socket.recv(1024).decode('utf-8')
                print(f"Received message: {message}")
            except:
                break

    def start(self):
        self.connect()
        threading.Thread(target=self.receive_messages).start()

if __name__ == "__main__":
    client = MessageQueueClient()
    client.start()

    # 模拟发送消息
    input("Press Enter to send a message to the server...")
    client.send_message("Hello, server!")