Das deutsche Python-Forum

Code: Alles auswählen

import socket
import threading
import multiprocessing
import time

class GameServer:

    def __init__(self):
        self.clients = {}  # dictionary to store client address -> socket connection
        self.game_state = {}  # dictionary to store game state

    def update_game_state(self, data, client_address):
        # Update game state based on data received
        # In a real game, there would be complex logic to update the game state
        # Here we just echo the received data
        self.game_state[client_address] = data

    def handle_client(self, client_socket, client_address):
        while True:
            data = client_socket.recv(1024)  # receive data from the client
            if not data:
                break

            # Update game state in a separate process
            p = multiprocessing.Process(target=self.update_game_state, args=(data, client_address))
            p.start()
            p.join()

            # Send updated game state to all clients
            for other_client_socket in self.clients.values():
                other_client_socket.sendall(data)

        client_socket.close()
        del self.clients[client_address]

    def run_server(self, host='localhost', port=12345):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
            server_socket.bind((host, port))
            server_socket.listen()

            while True:
                client_socket, client_address = server_socket.accept()
                self.clients[client_address] = client_socket
                threading.Thread(target=self.handle_client, args=(client_socket, client_address)).start()


# Run the server
game_server = GameServer()
game_server.run_server()

Code: Alles auswählen

import asyncio
import multiprocessing
from multiprocessing import Queue
import time


async def download_data(n):
    await asyncio.sleep(0.1)
    return f'Daten {n}'


def process_data(n, data, queue):
    time.sleep(5) 
    result = f'Verarbeitete {data}'
    queue.put(result)

async def producer(queue):
    for n in range(5):

        data = await download_data(n)

        process = multiprocessing.Process(target=process_data, args=(n, data, queue))
        process.start()

def consumer(queue):
    while True:
        result = queue.get() 
        if result == 'STOP':
            break
        print(result)

if __name__ == "__main__":
    queue = Queue()

    producer_process = multiprocessing.Process(target=asyncio.run, args=(producer(queue),))
    consumer_process = multiprocessing.Process(target=consumer, args=(queue,))
    producer_process.start()
    consumer_process.start()

    producer_process.join()
    queue.put('STOP')
    consumer_process.join()

Code: Alles auswählen

import asyncio
import multiprocessing
import websockets
import redis

import time, datetime, json, os, sys, uuid, pathlib, collections, logging

logging.root.setLevel(0)

redis_conn = redis.Redis(host='localhost', port=6379, db=0)

class WebSocketServer:
    def __init__(self, host, port, incoming_queue_name, outgoing_queue_name):
        self.host = host
        self.port = port
        self.incoming_queue_name = incoming_queue_name
        self.outgoing_queue_name = outgoing_queue_name
        self.active_connections = set()

    async def receive(self, websocket):
        async for message in websocket:
            start_time = datetime.datetime.now()
            redis_conn.rpush(self.incoming_queue_name, message)
            end_time = datetime.datetime.now()
            duration = (end_time - start_time).total_seconds()
            logging.info(f"Message received at {start_time}, processed in {duration} seconds")

    async def broadcast(self):
        while True:
            if redis_conn.llen(self.outgoing_queue_name) > 0:
                message = redis_conn.lpop(self.outgoing_queue_name).decode('utf-8')
                for websocket in self.active_connections:
                    if not websocket.closed:
                        await websocket.send(message)
            await asyncio.sleep(0.1)  # Avoid busy waiting

    async def server(self, websocket, path):
        # Add the new websocket connection to the set of active connections
        self.active_connections.add(websocket)
        try:
            receiver_task = asyncio.create_task(self.receive(websocket))
            broadcaster_task = asyncio.create_task(self.broadcast())  # This will send messages to all connected clients
            await asyncio.gather(receiver_task, broadcaster_task)
        finally:
            # Remove the websocket connection from the set of active connections
            self.active_connections.remove(websocket)


    async def start_server(self):
        async with websockets.serve(self.server, self.host, self.port, ssl=None):
            await asyncio.Future()  # This will keep the server running indefinitely.

    def run(self):
        asyncio.run(self.start_server())


class Message_Handler:
    def __init__(self, i, incoming_queue_name, outgoing_queue_name):
        self.i = i
        self.incoming_queue_name = incoming_queue_name
        self.outgoing_queue_name = outgoing_queue_name

    async def process_incoming_message(self):
        while True:
            message = redis_conn.lpop(self.incoming_queue_name)
            if message is not None:
                message = message.decode('utf-8')
                logging.info(f"Processed message: {message}")
                redis_conn.lpush(self.outgoing_queue_name, message.encode('utf-8'))
            else:
                await asyncio.sleep(self.i)

    def run(self):
        asyncio.run(self.process_incoming_message())


def main():
    incoming_queue_name = "incoming_messages"
    outgoing_queue_name = "outgoing_messages"

    ws_url = 'ws://example.com:1234'
    host = '0.0.0.0'  # Listen on all network interfaces
    port = 8765  # Replace this with your desired port number

    ws_server = WebSocketServer(host, port, incoming_queue_name, outgoing_queue_name)
    message_handler = Message_Handler(0.1, incoming_queue_name, outgoing_queue_name)

    process1 = multiprocessing.Process(target=ws_server.run)
    process2 = multiprocessing.Process(target=message_handler.run)

    process1.start()
    process2.start()

    process1.join()
    process2.join()


if __name__ == "__main__":
    main()