如何以高tps、多线程向rabbitmq发布消息答案

【问题标题】：How to publish messages to rabbitmq with high tps, multithreading如何以高tps、多线程向rabbitmq发布消息
【发布时间】：2018-07-21 19:57:30
【问题描述】：

由于通道不是线程安全的，我可以在发布之前同步通道实例，也可以在每次需要时创建一个通道并关闭它。

但在我看来，由于锁定或创建和销毁通道的成本，它们都没有良好的性能。

那么我应该如何以高tps向rabbitmq发布消息？有什么好的表扬吗？

【问题讨论】：

【解决方案1】：

你应该使用游泳池。

例如，使用Apache's Generic Object Pool 并提供打开、关闭和检查连接的实现。当你需要发布消息时，你从池中借用一个通道，使用它，然后返回它。

【讨论】：

【解决方案2】：

所以，第一件事。 通道不是连接。 在 RabbitMQ 中，通道与应用程序会话相同，而连接表示与服务器的底层 TCP 会话。 This answer 很好地解释了其中的一些内容。

TCP 会话的创建成本很高，因此它们的生命周期往往超出任何特定工作线程的生命周期。创建频道非常便宜 - 服务器所做的只是为您分配一个整数作为频道标识符，然后您就有了一个新频道。

RabbitMQ 上的大多数操作在失败时都会关闭通道。这样做是因为这样做没有实际后果。他们是否会关闭底层连接，这会给应用程序带来很多问题。

设计指导

如果您确实有很多处理正在进行，则池将适用于连接。关于如何做到这一点的讨论真的超出了我可以在简短回答中提供的范围。
池绝对不适合频道。通道是一种轻量级结构，旨在具有短暂的生命周期。如果它的持续时间超过一两次发布，那就太好了。但是你应该预料到，每次尝试操作时，都有可能会失败并关闭通道。这不会关闭底层连接，但必须重新建立一个新通道才能对代理执行任何操作。
消费者生命周期与渠道相关。当通道关闭时，附加的消费者也关闭。设计您的消费者对象（工作线程），使其能够在发生这种情况时获得连接并创建新通道，然后重新订阅自己。
避免跨线程共享通道。一个线程 = 一个通道。
如果您正在编写自己的协议实现库（不是必需的，但如果您需要细粒度控制也不错），分配一个线程来管理每个connection。不要并行读写 TCP 套接字，否则会破坏协议。
关于 Java 客户端，我认为您可以假设通道操作（读取和写入等）是非线程安全的，这就是您要坚持使用一个线程的原因/一个通道范式。我认为您可以假设创建通道（连接操作）是线程安全的。

【讨论】：

现在我有一个大于 2000 tps 的 pulish tps，实际上我正在创建频道并在每次需要发布时销毁它。但这意味着如果我想创建一个 chanel 并且每次都丢失它当我发布时，我需要与代理进行额外的两个 TCP 通信，这对代理来说是一个负担。现在我已经注意到代理已经在我的连接上触发了流控制，所以我现在正在尝试做任何事情来提高它的效率。我猜频道池是一种方式，不是吗？
没有。你有很多事情要做。您不需要在每次发布时都创建一个通道——一旦您的应用程序的发布组件有了一个通道，它就可以被重用，直到您做一些导致关闭的事情。每个连接的通道数没有限制。流量控制意味着您发布的速度对于另一端的消费者（接收消息的人）来说太快了，因此无论如何这将是一个全球性问题，但这些设置是可配置的。
1.但是通道不是线程安全的，所以我要么有多个通道，要么进行同步。所以总而言之，通道池或使用线程本地是一个好方法，不是吗？ 2. 流量控制是否受消耗率影响？是不是只受broker能力的影响（比如Kafka），所以你的意思是RabbitMQ不能在broker上发布海量消息消费慢？
如果您每次需要发布时都创建和销毁一个新线程，那么您遇到的问题可能比我能帮助您解决的问题更大。
线程由200个固定大小的固定trhead池管理