标签: message-queue

我期待在单个进程多线程应用程序上使用POSIX消息队列.mqueues将用于在线程之间共享数据.

我对它们在Linux内核中的工作方式感到有些困惑.是否所有消息都通过内核空间然后返回到接收上的用户空间？又来自用户空间线程,我做一个mq_send,消息最终在内核空间中,然后在接收时,它是从内核空间获取消息的另一个系统调用.如果是这样,对于高使用率的消息队列来说这不是非常低效吗？

我读了一篇关于多线程程序设计http://drdobbs.com/architecture-and-design/215900465的文章,它说这是"用异步消息替换共享数据的最佳实践.尽可能地保留每个线程的数据"隔离(非共享),让线程通过传递数据副本的异步消息进行通信".

令我困惑的是,我没有看到使用共享数据和消息队列之间的区别.我现在正在开发一个关于Windows的非gui项目,所以让我们使用windows的消息队列.并以传统的生产者 - 消费者问题为例.

使用共享数据,将有一个共享容器和一个锁定生产者线程和消费者线程之间容器的锁.当生产者输出产品时,它首先等待锁定然后向容器写入内容然后释放锁定.

使用消息队列,生产者可以简单地PostThreadMessage而不用阻塞.这是异步消息的优势.但我认为必须存在一些锁定两个线程之间的消息队列,否则数据肯定会被破坏.PostThreadMessage调用只是隐藏细节.我不知道我的猜测是否正确但是如果它是真的,优势似乎不再存在,因为两种方法都做同样的事情,唯一的区别是系统在使用消息队列时隐藏细节.

PS.也许消息队列使用非阻塞包含器,但我也可以使用前一种方式的并发容器.我想知道消息队列是如何实现的,这两种方式之间是否有任何性能差异？

更新:如果消息队列操作仍然在其他地方被阻止,我仍然没有得到异步消息的概念.如果我猜错了,请纠正我:当我们使用共享容器和锁时,我们将在我们自己的线程中阻塞.但是当使用消息队列时,我自己的线程立即返回,并将阻塞工作留给某个系统线程.

我们需要构建一个软件框架(或中间件),以便在单个机器上运行的不同软件组件(或模块)之间实现消息传递.该框架将提供以下功能:

• 模块之间的通信是通过"消息传递".
• 每个模块都有自己的消息队列和消息处理程序线程,它将同步处理每个传入消息.

根据上述要求,以下哪种方法是正确的(有其推理)？:

1. 将模块实现为进程,并通过共享内存进行消息传递
2. 在单个进程中将模块实现为线程,并通过将消息对象推送到目标模块的消息队列来进行消息传递.

有消息来源,有一些明显的缺点和优点:

• 在Option-2中,如果一个模块导致分段错误,则该过程(因此整个应用程序)将崩溃.并且一个模块可以直接访问/改变另一个模块的内存,这可能导致难以调试的运行时错误.
• 但是使用Option-1,您需要处理需要通信的模块刚刚崩溃的状态.如果软件中有N个模块,则系统中可能存在2 ^ N个活动/崩溃状态,这些状态会影响模块上运行的算法.
• 同样在Option-1中,发送方不能假设接收方已收到消息,因为它可能在那一刻崩溃了.(但是系统可以警告所有模块特定模块已经崩溃;这样,发送方可以断定接收方将无法处理该消息,即使它已成功接收到该消息)

我赞成Option-2,但我不确定我的论点是否足够稳固.你有什么看法？

编辑:根据要求澄清,这里有更多的规格细节:

• 这是一个将在Linux OS上运行的嵌入式应用程序.
• 不幸的是,我无法告诉你项目本身,但我可以说项目有多个组件,每个组件都将由自己的团队(3-4人)开发,并决定这些组件之间的通信组件/模块是通过某种消息传递框架.
• C/C++将用作编程语言.
• "模块接口API"将自动提供给模块开发人员的是:(1)消息/事件处理程序线程循环,(2)同步消息队列,(3)函数指针成员变量,您可以在其中设置消息处理函数.

我有一个场景,我想使用特定的路由键将一些消息发布到rabbitmq-exchange,例如.ABC

问题是可能已经存在任何已使用路由密钥"abc"绑定的队列,或者可能没有.这种情况的行为似乎是丢弃该消息,或者如果配置了死信交换,它将被路由到死信交换.

我想动态创建一个与路由密钥同名的队列,即"abc",如果该路由密钥没有队列,而不是丢弃或发送到DLX.

有没有任何已知的方法来做同样的事情？

我似乎无法在任何地方找到这些信息,或者我可能不理解它.在RabbitMQ的发布/订阅模式中,当生产者生成消息时,消费者如何知道队列中有新消息？

消费者是否经常轮询队列以检查是否有新消息或交换"推送"通知消费者是否有新消息？

我需要将在Heroku上运行的Node JS服务器移动到消息队列架构.目前,服务器接收HTTP请求,进行一些处理并进行响应.问题是处理需要一些时间,特别是当有大量请求时.这个漫长的处理时间会导致服务器超时,过载和崩溃!我的阅读告诉我需要后台工作人员来处理.

我对消息队列和后台工作者没有经验,我正在寻找一个非常简单的示例来开始.任何人都可以建议一个简单,易懂的模块或示例来开始？

我发现了一些例子,但它们很复杂,我迷路了!我想要一个我可以构建的准系统示例.

我的项目将Apache Kafka视为基于JMS的老化消息传递方法的潜在替代品.为了使这种转换尽可能顺利,如果替换排队系统(Kafka)具有异步订阅机制,这将是理想的,类似于我们当前项目使用MessageListener和MessageConsumer订阅主题并接收异步通知的JMS机制.如果Kafka没有严格遵守JMS API,我并不在乎,但相反,如果我不需要,我宁愿不重新设计我们的整套发布 - 订阅 - 通知类.

我可以找到各种KafkaConsumer 轮询 示例,但到目前为止还没有找到任何客户端通过异步通知通知新消息的示例.

有谁知道Kafka的当前版本(截至本文发布时为0.10.2)是否提供了这样的API,还是我一直试图使用轮询重写我的遗留代码？

我有一个RabbitMQ队列用于生成文档.基本上,每个文件都有type和state(新,处理,准备),所以我用像路由键的话题交换type.state.每次文档更改时,我都会将带有上一个文档描述的消息发送到交换机,并且它的工作状态非常好

但有时文档可以处理两次:

1. 用户发送新文件.所以新消息report.new被发送到交换.
2. 虽然worker尚未启动文档处理(队列尚未到达),但用户更新了文档.report.new发送同一文档的新消息.
3. 因此,现在工作人员得到第一条消息并开始他的工作,而文档被更改,因此这项工作完全没有意义.

现在我只是将小代码添加到工作程序中,将last_modified消息中的文档密钥与数据库中的密钥进行比较,如果它们不相同则查询消息.但是我不认为这是最好的解决方案.

我的想法是添加ID到消息头并有一些RabbitMQ插件,它将ID从队列中删除相同的旧消息.

谢谢.

PS也许另一个MQ引擎在这里有用吗？例如,ActiveMQ可能有这样的功能吗？

例如,我有一个博客基础上Django,我已经有几个功能,为用户login:edit_profile, ,share.

但现在我需要实施一个任务系统.

1. 用户登录,10 score每天奖励
2. 用户完成他的个人资料,奖励 20 score
3. 用户分享我的博客,奖励 30 score

我不想将奖励代码与普通功能代码混合使用.所以我决定使用message queue.伪代码可能如下所示:

@login_required
def edit_profile(request):
    user = request.user
    nickname = ...
    desc = ...
    user.save(...)
    action.send(sender='edit_profile', payload={'user_id': user.id})
    return Response(...)

奖励可以订阅此操作

@receiver('edit_profile')
def edit_profile_reward(payload):
    user_id = payload['user_id']
    user = User.objects.get(id=user_id)
    mission, created = Mission.objects.get_or_create(user=user, type='complete_profile')
    if created:
         user.score += 20
         user.save()

但我不知道这是不是正确的方法.如果是这样,我message queue应该使用什么？django-channel/ django-q或其他什么？如果没有,最佳做法是什么？

我想使用 Nats Jetstream 实现一次精确交付系统。文档称 Jetstream 有此选项，但没有有关其工作原理以及客户如何实现此选项的示例或详细信息。我知道在发布者端我们可以在创建Stream时设置MsgId并指定复制窗口，但是消费者端呢？