标签: message-queue

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

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

我们面临着使用IBM MQ over SFTP进行文件传输的选择.我听说过这种方法的优点,但我从未见过有人将它用于大文件.

所以主要问题:IMB MQ如何处理大文件的传输(最多100 MB)？它稳定吗？它是从大型机到UNIX服务器,如果它确实重要.

非常感谢.

简短介绍

我有一个基于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 …

我列出了每个队列中的队列和消息.以下是我的代码.但是,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更有可能成为完美的客户端解决方案.特别是在我的情况下,在用户付款后,他想立即收到硬币,而不是我们告诉他交易处理的情况,他的账户很快就会完成.在移动设备上具有响应式用户体验的优先级.你还在服务器上使用消息队列吗？

谢谢你的时间.

我们在开发实验室中设置了一个非常简单的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队列.我想保证每条消息都会消耗一次.我知道可以更改消息的可见性超时,例如,等于我的处理时间.如果我的进程花费的时间超过可见性超时(例如,连接速度慢),则其他线程可以使用相同的消息.

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

我正在使用Java和Cassandra的事件采购从头开始构建项目.我的应用程序基于微服务,在某些用例中,信息将以异步方式处理.我想知道Message Queue(例如Rabbit,Active MQ Artemis,Kafka等)将在这个环境中改进技术堆栈的哪个部分,如果我理解这些场景,我将不会使用它.

我需要帮助才能理解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 …

Tl;博士:"我如何通过一堆异步,无序的微服务推送消息,并知道该消息何时通过它们？"

我正在努力为特定的微服务架构找到合适的消息传递系统/协议.这不是"哪个是最好的"问题,而是关于我对设计模式/协议的选择的问题.

• 我在开始队列中有一条消息.让我们说一个带有序列化JSON的RabbitMQ消息
• 我需要该消息来通过任意数量的微服务
• 这些微服务中的每一个都是长期运行的,必须是独立的,并且可以以各种语言实现
• 消息所经过的服务顺序无关紧要.实际上,它不应该是同步的.
• 每个服务都可以将数据附加到原始消息,但其他服务会忽略该数据.应该没有合并冲突(每个服务写一个唯一的密钥).任何服务都不会改变或破坏数据.
• 一旦所有服务轮到他们,就应该使用原始数据和新数据将消息发布到第二个RabbitMQ队列.
• 微服务没有其他副作用.如果这些都在一个单片应用程序中(并且使用相同的语言),那么函数式编程将是完美的.

那么,问题是,通过各种服务管理该消息的适当方式是什么？我不想一次做一个,订单并不重要.但是,如果是这种情况,那么系统如何知道所有服务何时发生了重击并且最终消息可以写入结束队列(以使下一批服务完成它们).

我能想到的唯一,半优雅的解决方案是

1. 让第一个遇到消息的服务将该消息写入公共存储(比如mongodb)
2. 让每个服务做它的事情,标记它已完成该消息,然后检查所有服务是否已经轮到他们
3. 如果是这样,那最后一个服务将发布该消息

但是,这仍然需要每个服务都知道所有其他服务,并要求每个服务留下其标记.这些都不是必需的.

我对某种"牧羊人"服务持开放态度.

我很感激我错过的任何选择,并且愿意承认他们可能是一个更好的,基本的设计.

谢谢.