标签: message-queue

在Java Web应用程序中,我有一个类型为A的重复消息加载(例如,每小时20,000个).然后我有第二种类型的消息(类型B)偶尔显示但具有比类型A更高的优先级(例如,3,000).我希望能够使用开源软件在一台或多台机器上处理这些消息.

在我看来,如果我有一个基于优先级从其队列发送消息的JMS服务器,我可以用JMS做到这一点(例如,发送三个类型B的消息,然后发送一个类型A的消息,即使类型A的所有消息都是在消息队列的顶部).

您是否知道可以执行此操作的JMS服务器 - 或者您是否知道另一种实现方法？

我正在尝试编译用于在iPhone上使用的zeroMQ 的Objective-C绑定(https://github.com/jeremy-w/objc-zmq).

我已经添加了"objc-zmq"项目作为我想要包含它的项目的依赖项.但是,所有符号都可以找到,因为它没有链接到libzmq.a二进制文件.

如何编译libzmq.a以用于iPhone.此外,一旦我有了,我如何将我的项目链接到它？

我尝试了这篇文章中的一些建议:为iPhone编译C lib

然而,这些对我来说毫无结果.任何帮助将不胜感激.

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

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

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

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

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

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

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

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

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

我最近实现了一个自动回复包含任意主题标签的推文的系统.该系统包括定期抓取Twitter的流程和定期回复这些推文的流程.遵循我公司的传统,这些定期工作通过RDMS上的工作表实现,其具有状态列,其具有诸如"等待","处理"或"成功"的值.为了确保冗余,我通过利用低级别锁来运行多个相同的进程.

我的问题是,我正在周期性地实施RDMS中工作表的作业,这些作业通常是如何实现的.

当使用RabbitMQ作为Message Broker时,我有一个场景,其中多个并发使用者使用basic.get AMQP方法从队列中提取消息,并使用显式确认来从队列中删除消息.假设以下设置

Q具有消息M1,M2,M3并且消费者C1,C2和C3(每个都具有其自己的连接和信道)连接到它.

1. 如何在basic.get方法中处理并发？对basic.get方法的调用是否同步处理每个使用自己的连接和通道的并发消费者？C1,C2和C3发出basic.get调用以同时接收消息(假设服务器同时接收所有3个请求).

2. C1使用basic.get请求消息并获取M1.当C2请求消息时,由于它使用不同的连接,它是否再次获得M1？

3. 消费者如何以预定义的大小批量提取消息？

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

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

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

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

我正在尝试为现有队列编写使用者.

RabbbitMQ在一个单独的实例中运行,名为"org-queue"的队列已经创建并绑定到交换机.org-queue是一个持久队列,它还有一些额外的属性.

现在我需要从这个队列接收消息.我使用下面的代码来获取队列的实例

conn = Bunny.new
conn.start
ch = conn.create_channel    
q = ch.queue("org-queue")

它给我一个错误,说明不同的耐用性.默认情况下,Bunny使用的持久= false.所以我添加了持久的true作为参数.现在它说明了其他参数之间的区别.我是否需要指定所有参数才能连接到它？由于rabbitMQ由不同的环境维护,我很难获得所有属性.

有没有办法获取队列列表并在客户端中侦听所需的队列,而不是通过所有参数连接到队列.

我正在为消息队列编写一个lib.队列可以是Direct或Topic.Direct队列具有静态绑定密钥,而Topic队列可以具有动态密钥.

我想编写一个publish仅适用于Direct队列的函数.这有效:

{-# LANGUAGE DataKinds #-}

type Name = Text
type DirectKey = Text
type TopicKey = [Text]

data QueueType
  = Direct DirectKey
  | Topic TopicKey

data Queue (kind :: a -> QueueType)
  = Queue Name QueueType

这需要两个独立的构造函数

directQueue :: Name -> DirectKey -> Queue 'Direct

topicQueue :: Name -> TopicKey -> Queue 'Topic

但是当我去编写发布时,我需要匹配一个额外的模式,这是不可能的

publish :: Queue 'Direct -> IO ()
publish (Queue name (Direct key)) = …

我有一个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或其他什么？如果没有,最佳做法是什么？