Python中消息订阅应用开发的最优5个方案及代码实现

import redis
import threading

# 发布者
class RedisPublisher:
    def __init__(self, host='localhost', port=6379):
        self.redis_client = redis.Redis(host=host, port=port)

    def publish(self, topic, message):
        self.redis_client.publish(topic, message)
        print(f"Published message '{message}' to topic '{topic}'")

# 订阅者
class RedisSubscriber:
    def __init__(self, host='localhost', port=6379):
        self.redis_client = redis.Redis(host=host, port=port)
        self.pubsub = self.redis_client.pubsub()

    def subscribe(self, topic):
        self.pubsub.subscribe(topic)
        print(f"Subscribed to topic '{topic}'")

    def listen(self):
        for message in self.pubsub.listen():
            if message['type'] == 'message':
                print(f"Received message '{message['data']}' from topic '{message['channel']}'")

    def start_listening(self):
        threading.Thread(target=self.listen).start()

# 测试
if __name__ == "__main__":
    publisher = RedisPublisher()
    subscriber = RedisSubscriber()

    subscriber.subscribe('topic1')
    subscriber.start_listening()

    publisher.publish('topic1', 'Hello, Redis!')

import pika

# 生产者
def rabbitmq_producer():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='task_queue', durable=True)

    message = 'Hello, RabbitMQ!'
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=message,
        properties=pika.BasicProperties(delivery_mode=2)  # 消息持久化
    print(f"Sent message: {message}")
    connection.close()

# 消费者
def rabbitmq_consumer():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='task_queue', durable=True)

    def callback(ch, method, properties, body):
        print(f"Received message: {body}")
        ch.basic_ack(delivery_tag=method.delivery_tag)  # 消息确认

    channel.basic_consume(queue='task_queue', on_message_callback=callback)
    print("Waiting for messages...")
    channel.start_consuming()

# 测试
if __name__ == "__main__":
    rabbitmq_producer()
    rabbitmq_consumer()

from kafka import KafkaProducer, KafkaConsumer

# 生产者
def kafka_producer():
    producer = KafkaProducer(bootstrap_servers='localhost:9092')
    topic = 'test_topic'
    message = 'Hello, Kafka!'
    producer.send(topic, message.encode('utf-8'))
    producer.flush()
    print(f"Sent message: {message}")

# 消费者
def kafka_consumer():
    consumer = KafkaConsumer(
        'test_topic',
        bootstrap_servers='localhost:9092',
        auto_offset_reset='earliest',
        group_id='my_group'
    )
    print("Waiting for messages...")
    for message in consumer:
        print(f"Received message: {message.value.decode('utf-8')}")

# 测试
if __name__ == "__main__":
    kafka_producer()
    kafka_consumer()

import zmq

# 发布者
def zeromq_publisher():
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.bind("tcp://*:5555")

    topic = 'topic1'
    message = 'Hello, ZeroMQ!'
    socket.send_string(f"{topic} {message}")
    print(f"Sent message: {message}")

# 订阅者
def zeromq_subscriber():
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.connect("tcp://localhost:5555")
    socket.setsockopt_string(zmq.SUBSCRIBE, 'topic1')

    print("Waiting for messages...")
    while True:
        message = socket.recv_string()
        print(f"Received message: {message}")

# 测试
if __name__ == "__main__":
    import threading
    threading.Thread(target=zeromq_subscriber).start()
    zeromq_publisher()

import paho.mqtt.client as mqtt

# 发布者
def mqtt_publisher():
    client = mqtt.Client()
    client.connect("localhost", 1883, 60)

    topic = 'test/topic'
    message = 'Hello, MQTT!'
    client.publish(topic, message)
    print(f"Sent message: {message}")
    client.disconnect()

# 订阅者
def on_message(client, userdata, msg):
    print(f"Received message: {msg.payload.decode('utf-8')}")

def mqtt_subscriber():
    client = mqtt.Client()
    client.on_message = on_message
    client.connect("localhost", 1883, 60)
    client.subscribe("test/topic")
    print("Waiting for messages...")
    client.loop_forever()

# 测试
if __name__ == "__main__":
    mqtt_publisher()
    mqtt_subscriber()