标签: message-queue

我们的应用程序从DCOM(0x80010100)获取系统调用失败的RPC错误,我们怀疑目标线程的消息队列已满(尽管我不相信这是真的).我知道队列仅限于10,000消息,我想知道在常见情况下我们是否接近这个数字.有没有办法监控线程的消息队列的大小？

我找到的最有希望的选项是,GetQueueStatus但这不包括队列中只有其类型的消息数.

简短介绍

我有一个基于SEDA的系统,并使用MSMQ在不同的应用程序/服务之间进行通信(事件触发).

其中一个服务按文件获取消息,因此我有一个文件监听器,它读取文件内容并将其插入队列(或实际上是4个不同的队列,但这对第一个问题不是很重要).

服务器是Windows Server 2008

第一个问题 - 阅读速度慢下来

我在另一端读取这些消息的应用程序通常每秒从队列中读取大约20条消息,但是当发布消息的服务开始排队数千条消息时,读取​​将关闭,并且读取应用程序仅读取2-4条消息第二.当没有发布到队列时,读取的应用程序可以再次读取每秒最多20条消息.

阅读应用程序中的代码非常简单,在C#中开发,我在System.Messaging中使用Read(TimeSpan超时)函数.

问:为什么在队列中发布大量消息时读取速度会变慢？

第二个问题 - TPS的局限性

另一个问题是阅读本身.如果我使用1或5个线程从队列中读取,那么每秒可以读取的消息数似乎没有区别.我也尝试实现一个"循环解决方案",其中邮政服务发布到一组随机的4个队列,并且读取的应用程序有一个线程监听这些队列中的每一个,但即使我仍然只有20个TPS从1个队列中读取1个线程,1个队列,4个线程或4个队列(每个队列一个线程).

我知道线程中的处理大约需要50毫秒,所以如果当时只处理一条消息,那么20 TPS是完全正确的,但多线程的线索应该是并行处理消息而不是顺序消息.

服务器上有大约110个不同的队列.

问:为什么即使使用多线程和使用多个队列,我也无法在队列中收到超过20条消息？

这是今天运行的代码:

// There are 4 BackgroundWorkers running this function
void bw_DoWork(object sender, DoWorkEventArgs e) 
{
    using(var mq = new MessageQueue(".\\content"))
    {
        mq.Formatter = new BinaryMessageFormatter();

        // ShouldIRun is a bool set to false by OnStop()
        while(ShouldIRun)
        {
            try
            {
                using(var msg = mq.Receive(new TimeSpan(0,0,2))
                {
                    ProcessMessageBody(msg.Body); // This takes 50 ms to complete
                }
            }
            catch(MessageQueueException mqe)
            {
               // This …

好的,所以我对消息排队感兴趣.我已经对这个问题进行了大量的研究.我读了'编程windows azure'(关于Azure Queues),我阅读了大量有关Azure服务总线的教程和信息,我观看了关于消息模式的9频道视频等.

但我不明白的是:如何在现实场景中使用它？所有示例只是将一个字符串或一个包含一些数据的对象放入队列中,并从"另一侧"的队列中读取它.但是,您如何知道如何处理这些数据？例如:我可以尝试将Customer,Order和Address保存到数据库中,因此我将这3个对象放在队列中以便在另一侧读取它们将它们放入我的数据库中.我怎么知道如何处理这些物体？

我只有一些问题:

1. 将哪种数据放入队列中.可能是一个命令,它在读取时执行,因此队列中的所有内容都具有相同的接口,并且对象本身会弄清楚该怎么做？
2. 在什么层之间使用队列？我正考虑将内容放入服务层的队列中(在验证之后)并在数据访问层中读取它,以便将其放入数据库中.
3. 应该有多少队列？您想要连接的图层之间只有一个队列,或者它们之间有多个队列(可能出于不同的目的,虽然我想不出任何其他目的)？
4. 这种松散耦合形式允许我们对请求进行排队,以便稍后处理它们(例如:当我们想要重新启动数据库时).这很酷但是如果我想读取数据而不是写它呢？然后数据库应该在线.我应该通过队列读取数据还是我的服务层只是从数据访问层提取数据并将其传递给表示层？

我认为这是我头脑中嗡嗡作响的大部分问题.我希望有人能为我解决这个问题.

我列出了每个队列中的队列和消息.以下是我的代码.但是,QueueBrowser无法正确检索邮件.

比方说,我有一个名为TestQueue的队列,它有1000条消息.

我第一次运行我的程序时它只显示200条消息.第二 - 400第三 - 600第四 - 800第五 - 1000

你能告诉我如何解决这个问题吗？

ConnectionFactory out = new ActiveMQConnectionFactory("tcp://localhost:61616?jms.prefetchPolicy.all=10000");
ActiveMQConnection connection = (ActiveMQConnection) out.createConnection();

connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

Set<ActiveMQQueue> amqs = connection.getDestinationSource().getQueues(); 
Iterator<ActiveMQQueue> queues = amqs.iterator(); 

while ( queues.hasNext() ) 
{ 
    ActiveMQQueue queue_t = aqueues.next(); 
    String q_name = queue_t.getPhysicalName();
    List<ActiveMQMessage> msgList = ((ActiveMQSession) session).getUnconsumedMessages();

    System.out.println( "\nQueue = " + q_name);

    QueueBrowser queueBrowser = session.createBrowser(queue_t);
    Enumeration e = queueBrowser.getEnumeration();

    int numMsgs = 0;
    while(e.hasMoreElements()) 
    {
         Message message = (Message) e.nextElement();
         numMsgs++; …

我刚用IAP制作iOS应用程序来购买虚拟硬币(耗材).后端是Ruby on Rails.我现在对整个解决方案有一些担忧.

我目前的IAP流程描述如下:

• 第1步 - iOS客户端向我们的后端发送请求并获取产品ID.
• 第2步 - 客户端通过Store Kit检索产品列表,其中包含步骤1中的ID
• 第3步 - 用户点击购买,客户端向Store Kit发送付款请求
• 第4步 - 完成交易后,客户将收据令牌发送到我们的后端
• 步骤5 - 服务器首先验证收据,以确保它是有效的验证
• 第6步 - 服务器调用App Store验证收据并处理交换(DB中的Recharge帐户)
• 第7步 - 客户端获取响应并显示帐户余额

我以前从事过电子商务/支付解决方案,而IAP则采取其他方式,但通常支付网关会将收据服务器发送到服务器进行验证.我们总是有消息队列排序解决方案,以确保所有事务都是稳定和可维护的.

我的问题是:

1. 验证IAP收据的最佳做法是什么？如何避免来自我们服务器的验证中断？(例如,用户花钱并在客户端上完成交易,但在验证服务器上的收据时失败.没有IAP恢复事务支持.)

2. Apple IAP更有可能成为完美的客户端解决方案.特别是在我的情况下,在用户付款后,他想立即收到硬币,而不是我们告诉他交易处理的情况,他的账户很快就会完成.在移动设备上具有响应式用户体验的优先级.你还在服务器上使用消息队列吗？

谢谢你的时间.

Azure Service Bus中死信队列中的邮件是否过期？

一些解释

我有这些队列设置:

var queueDescription = new QueueDescription("MyTestQueue")
{
    RequiresSession = false,
    DefaultMessageTimeToLive = TimeSpan.FromMinutes(1),
    EnableDeadLetteringOnMessageExpiration = true,
    MaxDeliveryCount = 10
};

namespaceManager.CreateQueue(queueDescription);

当我将一些消息放入Azure Service Bus消息队列(而不是来自Azure存储的队列)并且不消耗它们时(永远),它们将自动移动到死信队列.

但是,如果我也没有死信队列的消费者,这些消息是否会从死信队列中删除,还是会永远留在那里？(是否有一些官方文件说明这应该如何工作？)

我的试炼

在我的试验中,我在队列中放置了3条消息.他们在2分钟左右就死了.他们至少在一天内留在死信队列中,并没有被删除.

虽然调用NamespaceManager.GetQueueAsync()给了我上面的值(注意如何MessageCount仍然3但DeadLetterMessageCount很奇怪0),我仍然可以从死信队列中收到消息.(所以他们没有从队列中删除.)

我有以下配置:

<bean id="mongoDbMessageStore" class="org.springframework.integration.mongodb.store.MongoDbMessageStore">
    <constructor-arg ref="mongoDbFactoryDefault"/>
</bean>

<!-- the queue capacity is unbounded as it uses a persistent store-->
<int:channel id="logEntryChannel">
    <int:queue message-store="mongoDbMessageStore"/>
</int:channel>

<!-- the poller will process 10 messages every 6 seconds -->
<int:outbound-channel-adapter channel="logEntryChannel" ref="logEntryPostProcessorReceiver" method="handleMessage">
    <int:poller max-messages-per-poll="10" fixed-rate="6000"/>
</int:outbound-channel-adapter>

并将消息处理程序定义为

@Override
public void handleMessage(Message<?> message) throws MessagingException {
    Object payload = message.getPayload();
    if (payload instanceof LogEntry) {
        LogEntry logEntry = (LogEntry) payload;
        String app = (String) message.getHeaders().get("app");
        logger.info("LogEntry Received - " + app + " " …

我们在开发实验室中设置了一个非常简单的Kafka 0.8.1.1.它只是一个节点.我们会定期遇到这个错误:

[2015-08-10 13:45:52,405] ERROR Controller 0 epoch 488 initiated state change for partition [test-data,1] from OfflinePartition to OnlinePartition failed (state.change.logger)
kafka.common.NoReplicaOnlineException: No replica for partition [test-data,1] is alive. Live brokers are: [Set()], Assigned replicas are: [List(0)]
        at kafka.controller.OfflinePartitionLeaderSelector.selectLeader(PartitionLeaderSelector.scala:61)
        at kafka.controller.PartitionStateMachine.electLeaderForPartition(PartitionStateMachine.scala:336)
        at kafka.controller.PartitionStateMachine.kafka$controller$PartitionStateMachine$$handleStateChange(PartitionStateMachine.scala:185)
        at kafka.controller.PartitionStateMachine$$anonfun$triggerOnlinePartitionStateChange$3.apply(PartitionStateMachine.scala:99)
        at kafka.controller.PartitionStateMachine$$anonfun$triggerOnlinePartitionStateChange$3.apply(PartitionStateMachine.scala:96)
        at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:743)

任何人都可以推荐一种从中恢复的策略吗？有这样的事情还是我们需要构建另一个或两个节点并在我们的主题上设置复制因子来覆盖我们放入集群的所有节点？

我们有3个zookeeper节点对Storm和HBase等其他应用程序响应非常好,因此我们非常有信心ZooKeeper不应该归咎于此.有任何想法吗？

我知道可以使用多个线程来使用SQS队列.我想保证每条消息都会消耗一次.我知道可以更改消息的可见性超时,例如,等于我的处理时间.如果我的进程花费的时间超过可见性超时(例如,连接速度慢),则其他线程可以使用相同的消息.

保证邮件处理一次的最佳方法是什么？

我需要帮助才能理解multiprocessing.Queue.我遇到的问题是,queue.get(...)与调用queue.put(...)和队列缓冲区(deque)相比,获得结果是非常有趣的.

这种泄漏抽象导致我调查队列的内部.它直截了当的源代码只是指向deque实现,这似乎也很简单,我不能用它来解释我所看到的行为.另外我读到Queue使用管道,但我似乎无法在源代码中找到它.

我把它归结为一个重现问题的最小例子,我在下面指定了一个可能的输出.

import threading
import multiprocessing
import queue

q = None
def enqueue(item):
    global q
    if q is None:
        q = multiprocessing.Queue()
        process = threading.Thread(target=worker, args=(q,))  # or multiprocessing.Process Doesn't matter
        process.start()
    q.put(item)
    print(f'len putted item: {len(item)}. qsize: {q.qsize()}. buffer len: {len(q._buffer)}')


def worker(local_queue):
    while True:
        try:
            while True:  # get all items
                item = local_queue.get(block=False)
                print(f'len got item: {len(item)}. qsize: {q.qsize()}. buffer len: {len(q._buffer)}')
        except queue.Empty:
            print('empty')


if …