保留消息直到使用 Python + Stomp 进行订阅

Question

我目前正在编写两个脚本来订阅消息服务器，使用 stomp 客户端库，write.py 写入数据和 read.py em> 获取数据。

如果我先启动 read.py 然后运行 ​​write.py，write.py 会正确接收消息。

但是，如果我先运行 write.py 然后再运行 read.py，read.py 不会检索之前发送的任何消息到服务器。

以下是脚本的相关部分。

如何实现 write.py 放入队列的消息在 read.py 订阅和检索它们之前一直保留？

write.py

def writeMQ(msg):
    queue = '/topic/test'
    conn = stomp.Connection(host_and_ports=[(MQ_SERVER, MQ_PORT)])
    try:
        conn.start()
        conn.connect(MQ_USER, MQ_PASSWD, wait=True)
        conn.send(body=msg, destination=queue, persistent=True)
    except:
        traceback.print_exc()
    finally:
        conn.disconnect()

    return

read.py

class MyListener(stomp.ConnectionListener):    
    def on_error(self, headers, message):    
        print ('received an error {0}'.format(message))    

    def on_message(self, headers, message):    
        print ('received an message {0}'.format(message))    


def readMQ():                                                
    queue = '/topic/test'     
    conn = stomp.Connection(host_and_ports=[(MQ_SERVER, MQ_PORT)])    
    try:                                                              
        conn.set_listener("", MyListener())                           
        conn.start()                                                  
        conn.connect(MQ_USER, MQ_PASSWD, wait=True)    

        conn.subscribe(destination=queue, ack="auto", id=1)                       


        stop = raw_input()                  
    except:                   
        traceback.print_exc()    
    finally:                     
        conn.disconnect()        

    return                   

Answer 1

问题是消息被发送到一个主题。

Apollo Documentation 描述主题和队列的区别如下：

即使没有附加订阅，队列也会保留未使用的消息，而当没有连接的订阅时，主题会丢弃消息。

因此，当 read.py 首先启动并监听时，主题识别订阅并转发消息。但是当 write.py 首先是 startet 时，消息会被丢弃，因为没有订阅的客户端。

因此您可以使用队列而不是主题。如果服务器能够静默创建队列，只需设置

queue = '/queue/test' .

不知道用的是哪个版本的stomp，但是找不到参数

send(..., persistent=True) .

无论如何，持久化不是正确的方法，因为它仍然不允许简单地保留消息以供以后连接，但会保存消息以防服务器故障。

你可以使用

retain:set

改为主题消息的标题。