标签: message-queue

我在玩 POSIX 队列，但遇到了一个问题。创建新队列时，我可以指定例如消息的大小以及队列中可以有多少消息。我的正常限制是 10，见

/proc/sys/fs/mqueue/msg_max

有没有一种简单的方法可以在程序执行期间更改它，除了

echo number > /proc/sys/fs/mqueue/msg_max

也许存在一些设置这些东西的系统调用。

在 Microsoft Azure 中，在服务总线下，您可以看到不同队列/主题等的活动图。

该图显示了各种不同的行，例如传入消息、传出消息、成功请求等。

什么是请求？什么是Message？

请求似乎是不断发生的，并且通常比消息的数量（以主题为单位）大得多。即使侦听队列的服务器数量保持不变 - 有时请求数量会急剧增加而消息数量没有任何变化。

我正在尝试定期检查 Kafka 中某个主题的队列大小。虽然，我什至不知道如何检查该指标甚至一次。我对 Kafka 完全陌生，所以我不确定该怎么做。我认为这将涉及创建生产者或消费者来与队列进行交互，但我遇到了障碍。

我正在研究什么基本上是一个高度可用的分布式消息传递系统。系统通过 HTTP 或 TCP 从某处接收消息，对其执行各种转换，然后将其发送到一个或多个目的地（也使用 TCP/HTTP）。

系统要求发送到给定目的地的所有消息都是有序的，因为有些消息建立在先前消息的内容之上。这限制了我们按顺序处理消息，每条消息大约需要 750 毫秒。因此，例如，如果有人每 250 毫秒向我们发送一条消息，我们将被迫将消息排在彼此后面。这最终会在高负载下在消息处理中引入无法容忍的延迟，因为每条消息在轮到它之前可能必须等待数百条其他消息被处理。

为了解决这个问题，我希望能够在不违反我们按顺序发送它们的要求的情况下并行化我们的消息处理。

我们可以轻松地横向扩展我们的处理。缺失的部分是一种确保即使消息被乱序处理的方法，它们也会被“重新排序”并按照接收顺序发送到目的地。我正在努力寻找实现这一目标的最佳方式。

Apache Camel 有一个叫做 Resequencer 的东西可以做到这一点，它包括一个很好的图表（我没有足够的代表直接嵌入）。这正是我想要的：接收无序消息并将它们按顺序排列的东西。

但是，我不希望它是用 Java 编写的，我需要该解决方案具有高可用性（即能够抵抗典型的系统故障，如崩溃或系统重新启动），我认为 Apache Camel 无法提供。

我们的应用程序是用 Node.js 编写的，使用 Redis 和 Postgresql 进行数据持久化。我们将Kue库用于我们的消息队列。尽管 Kue 提供优先排队，但功能集对于上述用例来说太有限了，所以我认为我们需要一种替代技术来与 Kue 协同工作来重新排序我们的消息。

我试图在网上研究这个话题，但我找不到预期的那么多信息。看起来像分布式架构模式的类型，会有大量的文章和实现，但我没有看到那么多。搜索诸如“消息重新排序”、“乱序处理”、“并行消息处理”等内容，可以找到大多数只是放宽基于分区或主题或诸如此类的“有序”要求的解决方案。或者，他们谈论在单台机器上的并行化。我需要一个解决方案：

• 可以以任何顺序同时处理多条消息。
• 将始终按照消息到达系统的顺序发送消息，无论它们以何种顺序进行处理。
• 可从 Node.js 使用
• 可以在 HA 环境中运行（即它的多个实例同时运行在同一个消息队列上，没有不一致的情况。）

我们目前的计划，对我来说很有意义，但我在网上找不到任何描述，是使用 Redis 来维护一组进行中和准备发送的消息，按到达时间排序。粗略地说，它是这样工作的：

1. 当收到一条消息时，该消息被放入进行中的集合中。
2. 当消息处理完成时，该消息被放入准备发送集。
3. 只要在进行中和准备发送集的前面都有相同的消息，就可以发送该消息并且它会按顺序进行。

我会编写一个小型 Node 库，它使用原子 Redis 事务通过优先级队列式 API 实现此行为。但这只是我自己想出来的，所以我想知道：是否有其他技术（理想情况下使用我们已经在使用的 Node/Redis 堆栈）可以解决重新排序乱序消息的问题? 或者是否有其他术语可以用作我可以用作研究关键字的此问题？谢谢你的帮助！

我正在使用 AWS sdk 开发基于 SQS 的队列发布订阅。

在非常基本的原型中，我将消息同时推送到我已经创建的队列中。但是，我如何确定我的消息被推送到队列中？

无论出于何种原因，推送的消息在此过程中都不会失败的保证是什么？

Amazon API 查询提供具有与该特定发送消息请求关联的 RequestID 的 ResponseMetadata。得到它是否足以肯定地确认将消息推送到 Queue ？

String requestId = sqs.sendMessage(request).getSdkResponseMetadata().getRequestId();

Amazon sendmessage 请求是否没有提供带有 Http 代码的响应正文？

由于 celery 的可靠性和调度问题，我们决定评估替代方案。我一直在努力在两个消息队列解决方案之间建立一个关于基本性能的基准。

我目前的方法是在两个不同的队列中放置 1000 个任务（获取 nvie.com 并计算网站上的字数），并测量 4 celery（20 秒）与 4 rq 工人（70 秒）的速度。我的代码是https://github.com/swarchris8/celery-vs-rq-benchmark我通过命令行运行 celery，通过 Mac 上的主管运行 rq，从 vagrant 文件中可以清楚地看到 rq 的 Ubuntu 运行指令。

Celery 的性能要好得多，我不确定我测量队列清除速度的测试设置是否存在用于测量任务吞吐量的缺陷。我也在使用默认的 RQ 工人，我怀疑它可能会慢得多。

我的方法是在吞吐量方面对两个消息队列系统进行基准测试的正确方法吗？你采取了哪些方法？celery 比 RQ 快这么多吗？

我们在不同的 VM 上有 2 个 ActiveMQ 节点（例如主机：amq1、amq2）。它们作为主/从集群链接。

我们想使用故障转移协议连接到这个集群。如何才能做到这一点？春季启动配置：

@Configuration
@EnableWebSocketMessageBroker
public class WebsocketConfiguration extends AbstractWebSocketMessageBrokerConfigurer {

  @Override
  public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableStompBrokerRelay("/topic")
        .setRelayHost(activeMQProperties.getRelayHost())
        .setRelayPort(activeMQProperties.getRelayPort());
  }
}

不幸的是，这里我们只能设置一台主机和一个端口。我们怎么能设置这样的东西：

failover:(stomp://amq1:61613,stomp://amq2:61613)

更新：目前使用的是 Spring Boot 2.3.5

谁能解释一下Android中的消息队列是什么？它是正在运行的进程列表吗？我找不到很好的来源来解释它。

我问这个问题是因为我正在阅读有关视图类的方法帖子。

邮政

添加到 API 级别 1 布尔值帖子（可运行操作）

导致 Runnable 添加到消息队列中。可运行程序将在用户界面线程上运行。

先感谢您。

在python中，在两个s之间建立单向通信的惯用方法是什么threading.Thread，将它们称为 threada和 thread b。

a是生产者，它不断地产生价值以供b消费。

b是消费者，它读取 生成的一个值a，用协程处理该值，然后读取下一个值，依此类推。

插图：

q = very_magic_queue.Queue()


def worker_of_a(q):
    while True:
        q.put(1)
        time.sleep(1)

a = threading.Thread(worker_of_a, args=(q,))
a.start()


async def loop(q):
    while True:
        # v must be processed in the same order as they are produced
        v = await q.get()
        print(v)

async def foo():
    pass

async def b_main(q):
    loop_fut = asyncio.ensure_future(loop(q))
    foo_fut = asyncio.ensure_future(foo())
    _ = await asyncio.wait([loop_fut, foo_fut], ...)
    # blah blah blah

def …

当我查看机器人操作系统（ROS）的教程时，我发现大多数示例代码将发布者的队列大小设置为较大的值，例如1000。我认为这会导致节点失去实时响应。

人们出于什么目的将其设置为那么大的值？