Ereignisse mit PubSub veröffentlichen und abonnieren

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# publisher.py
import zmq
import time
import random

def run_publisher():
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.bind("tcp://*:5555")  # Binds to all network interfaces on port 5555
    
    print("Publisher started. Publishing messages...")
    
    # Simple message generator
    topics = ["weather", "traffic", "news"]
    
    try:
        while True:
            topic = random.choice(topics)
            message = f"Update for {topic}: {time.time()}"
            print(f"Sending: {topic} {message}")
            
            # Format: topic message
            socket.send_string(f"{topic} {message}")
            
            time.sleep(1)
    except KeyboardInterrupt:
        print("Publisher shutting down")
        socket.close()
        context.term()

if __name__ == "__main__":
    run_publisher()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# subscriber.py
import zmq
import sys

def run_subscriber(topics=None):
    # Default to all topics if none specified
    if topics is None:
        topics = ["weather", "traffic", "news"]
    
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    
    # Connect to publisher
    #socket.connect("tcp://localhost:5555")  # For local testing
    socket.connect("tcp://publisher_ip:5555")  # For network testing
    
    # Subscribe to topics
    for topic in topics:
        socket.setsockopt_string(zmq.SUBSCRIBE, topic)
    
    print(f"Subscriber started. Listening for topics: {topics}")
    
    try:
        while True:
            message = socket.recv_string()
            topic, data = message.split(' ', 1)
            print(f"Received [{topic}]: {data}")
    except KeyboardInterrupt:
        print("Subscriber shutting down")
        socket.close()
        context.term()

if __name__ == "__main__":
    # If topics provided as command line arguments, use those
    topics = sys.argv[1:] if len(sys.argv) > 1 else None
    run_subscriber(topics)