【问题标题】:rabbitmq is it possible to prevent buggy client to create too many temp queues?rabbitmq 是否可以防止错误的客户端创建过多的临时队列?
【发布时间】:2013-05-20 14:06:12
【问题描述】:

如果客户端配置创建临时队列来接收消息,是否可以配置只允许创建一个队列? 如果有问题的客户端在服务器上创建了太多队列,对服务器来说会是一个大问题吗?

如何配置防止客户端浪费资源?

-module(amqp_example).

-include("amqp_client.hrl").

-compile([export_all]).

test() ->
    %% Start a network connection
    {ok, Connection} = amqp_connection:start(#amqp_params_network{}),
    %% Open a channel on the connection
    {ok, Channel} = amqp_connection:open_channel(Connection),

    %% Declare a queue
    #'queue.declare_ok'{queue = Q}
        = amqp_channel:call(Channel, #'queue.declare'{}),
    ok = create_temp_queue(Channel,10),

    %% Publish a message
    Payload = <<"foobar">>,
    Publish = #'basic.publish'{exchange = <<>>, routing_key = Q},
    amqp_channel:cast(Channel, Publish, #amqp_msg{payload = Payload}),

    %% Get the message back from the queue
    Get = #'basic.get'{queue = Q},
    {#'basic.get_ok'{delivery_tag = Tag}, _Content}
        = amqp_channel:call(Channel, Get),

    %% Do something with the message payload
    %% (some work here)

    %% Ack the message
    amqp_channel:cast(Channel, #'basic.ack'{delivery_tag = Tag}),

    %% Close the channel
    amqp_channel:close(Channel),
    %% Close the connection
    amqp_connection:close(Connection),

    ok.

create_temp_queue(Channel,Loop)->
    [     %% Declare a queue
          #'queue.declare_ok'{queue = _Q}
          = amqp_channel:call(Channel, #'queue.declare'{})
          ||
        _X <- lists:seq(1,Loop)].


(emacs@yus-iMac.local)58> amqp_example:test().
** exception error: no match of right hand side value 
                    [{'queue.declare_ok',
                         <<"amq.gen-AqAaMLydgMf43y_XoYSdq5">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-A75g--nsvheNbwYMr34M-E">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-wmkOrALHBIj6Ot6ZuZZOQJ">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-wX2NmwMHBeDaKLvoZgJhEh">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-gcvScDp-RFMVwxWpyWjI-9">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-Q4CS7jNu3cde0RNdVdO3PJ">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-QKNrG8IJPVvfAlLukq38x_">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-wqJ2V1HQDaJjOzRDhv8gT4">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-AYeZiuNYsFOUMVw6xKcZh4">>,0,0},
                     {'queue.declare_ok',
                         <<"amq.gen-AJDqT2h2fq9cZOsVbNESi0">>,0,0}]
     in function  amqp_example:test/0 (src/amqp_example.erl, line 16)


yus-iMac:~ yuchen$ sudo rabbitmqctl list_queues
Password:
Listing queues ...
amq.gen-Q4CS7jNu3cde0RNdVdO3PJ  0
amq.gen-QKNrG8IJPVvfAlLukq38x_  0
amq.gen-AqAaMLydgMf43y_XoYSdq5  0
amq.gen-AJDqT2h2fq9cZOsVbNESi0  0
amq.gen-wqJ2V1HQDaJjOzRDhv8gT4  0
amq.gen-AYeZiuNYsFOUMVw6xKcZh4  0
amq.gen-wzvWzxXo2MJVZsyrwfzM8A  0
amq.gen-A75g--nsvheNbwYMr34M-E  0
amq.gen-gcvScDp-RFMVwxWpyWjI-9  0
amq.gen-wX2NmwMHBeDaKLvoZgJhEh  0
 amq.gen-wmkOrALHBIj6Ot6ZuZZOQJ 0
 ...done.

【问题讨论】:

  • 这取决于你在做什么。你为什么要编写代码让你的客户只创建这样的对象?可以检查交换是否存在,甚至可以检查队列是否存在,甚至可以检查队列是否绑定到交换!你甚至可以创建一个绑定到指定交换的持久队列。

标签: erlang rabbitmq


【解决方案1】:

客户端应创建 1 个具有特定名称的队列,并确保将其设置为自动删除。这样,客户端将不会重新创建预先存在的队列。每个客户端都会创建自己的特定队列,因为每个客户端都会相应地为其队列命名。

【讨论】:

  • 我的意思是错误客户端。它不会做相应的。所以需要限制。
  • 我不明白,你说的bug客户端是什么意思?
  • 我在上面添加了代码。另外,如果使用命名队列,每个知道队列名称的恶意客户端都可以从队列中订阅消息,给系统带来麻烦。
【解决方案2】:

好的。现在让我们假设您已经正确设置了Erlang Rabbitmq client。使用用户指南 here

1.通用 RABBITMQ 连接器

-模块(rabbit_utils)。 -编译(export_all)。 -定义(兔子服务器,“本地主机”)。 -定义(兔子端口,9001)。 -记录(管道,{连接,通道})。 创建管道()-> 尝试 amqp_connection:start(#amqp_params_network{host = ?RABBIT_SERVER, 端口 = RABBIT_PORT}) 的 {好的,连接}-> 试试 amqp_connection:open_channel(Connection) 的 {好的,频道} -> #管道{ 连接=连接, 频道 = 频道 }; _ -> amqp_connection:关闭(连接),错误 抓住 _:_ -> amqp_connection:close(Connection),错误 结尾; _ -> 错误 抓住 _:_ -> 错误 结尾。 close_pipe(通道,连接)-> 尝试 amqp_channel:close(Channel) of _ -> ok catch _:_ -> ok end, 尝试 amqp_connection:close(Connection) of _ -> ok catch _:_ -> ok end, 好的。

随着我们的继续,您会注意到,我一直在创建管道和关闭管道。这是因为,在最新版本的erlang rabbitmq client 中,连接和通道变量是 Erlang 进程。每当这个库中的某些事情没有按计划进行时,这些进程就会在运行时终止。出于这个原因,我将大部分代码困在try ... catch ... end 代码块中。如果您正在使用gen_servers,则将它们设为trap_exit = true,然后将它们链接到RabbitMQ 客户端进程(连接和通道)。因此,每当连接中断时,Connection 进程都会死掉,将其捕获在您的 gen_server 中,然后尝试另一个连接,如果失败,也许您有许多 RABBITMQ 服务器可以通过 fail over 连接。概念在这里很深,但让我们继续。


2.交换管理器
这个确保交换存在于RABBITMQ 服务器上。交换必须是binary。我假设我们需要持久的交换和队列。持久交换和队列的一个优点是您只需创建一次,即使 RABBITMQ 服务重新启动,它也会使用这些交换和队列启动。

确保交换存在(交换)-> 案例 create_pipe() 的 错误->错误; #pipe{connection = Connection,channel = Channel} -> RandomQueue = list_to_binary(guid()), #'queue.declare_ok'{} = amqp_channel:call(Channel,#'queue.declare'{queue = RandomQueue}), 绑定 = #'queue.bind'{queue = RandomQueue,exchange = Exchange, routing_key = >}, 试试 amqp_channel:call(Channel, Binding) 的 #'queue.bind_ok'{} -> 删除 = #'queue.delete'{queue = RandomQueue}, #'queue.delete_ok'{} = amqp_channel:call(Channel, Delete), close_pipe(通道,连接), 好的; _ -> close_pipe(通道,连接), create_exchange(交换,服务器) 抓住 _R:_T -> close_pipe(通道,连接), create_exchange(交换,服务器) 结尾; _ -> 错误 结尾。 create_exchange(交换)-> 案例 create_pipe() 的 错误->错误; #pipe{connection = Connection,channel = Channel} -> Exc = #'exchange.declare'{exchange = Exchange,durable = true}, 尝试 amqp_channel:call(Channel,Exc) 的 #'exchange.declare_ok'{} -> 好的; _ -> close_pipe(通道,连接),错误 抓住 _:_ -> close_pipe(通道,连接),错误 结尾 结尾。 引导()-> 随机:种子(现在()), MD5 = erlang:md5(term_to_binary([random:uniform(7677771995517),{self(),time(),node(), now(), make_ref()}])), MD5List = 列表:nthtail(2, binary_to_list(MD5)), F = fun(N) -> f("~2.16.0B", [N]) 结束, L = [F(N) || N f(S,[])。 f(S,Args) -> 列表:flatten(io_lib:format(S,Args))。


3.队列管理器
对于大型系统,我们发现使用一个交换器(或几个交换器)更容易,但队列数量与系统交互所需的数量一样多。想象一下,在集群中,每个系统都有一个队列,当其他系统需要向该系统询问某些内容时,它们会在其中发送消息。此外,当一个系统收到另一个系统的请求时,它会通过将答案发送到该系统的队列中来回复该请求。使用良好的消息格式协议,它可以实现系统识别、区分请求和回复、时间戳、系统 ping、远程过程调用等,您最终会在集群中使用消息路由器。在最近的一个项目中,我们提出了一个远程过程协议,使用JSON 作为消息格式。系统可以通过 AMQP 接口调用方法并将参数传递给远程系统。我无法在这里详细介绍,但只知道,它是如此 Genuis 我们现在依赖它 100%

无论如何,我们首先需要确保存在队列,然后,如果是,检查它是否绑定到我们的交易所。我们走吧 !!

ensure_queue(Queue,Exchange)->
#pipe{connection = Connection,channel = Channel} = create_pipe(), Get = #'basic.get'{queue = Queue}, 当然 = 尝试 amqp_channel:call(Channel, Get) of #'basic.get_empty'{} -> queue_exists;
{#'basic.get_ok'{}, _Content} -> queue_exists; _ -> queue_missing 抓住 _EE:_EE2 -> queue_missing 结尾, close_pipe(通道,连接), 肯定的 queue_exists -> ensure_bound(Queue,Exchange); queue_missing -> create_queue_and_ensure_bound(队列,交换) 结束。
ensure_bound(Queue,Exchange)->
#pipe{connection = Connection,channel = Channel} = create_pipe(), B1 = #'queue.bind'{queue = Queue,exchange = Exchange,routing_key = Queue}, 尝试 amqp_channel:call(Channel, B1) of #'queue.bind_ok'{} -> close_pipe(通道,连接), 好的; _OtherErr -> 错误 抓住 _ET:_EY -> 错误 结束。

create_queue_and_ensure_bound(队列,交换)-> #pipe{connection = Connection,channel = Channel} = create_pipe(), TheQ = #'queue.declare'{queue = Queue,durable = true}, 尝试 amqp_channel:call(Channel,TheQ) 的 #'queue.declare_ok'{} -> 绑定 = #'queue.bind'{queue = Queue,exchange = Exchange,routing_key = Queue}, 试试 amqp_channel:call(Channel, Binding) 的 #'queue.bind_ok'{} -> close_pipe(通道,连接), 错误; _其他2 -> close_pipe(通道,连接), 错误 抓住 _ER11:_ER22 -> close_pipe(通道,连接), 错误 结尾; _其他-> close_pipe(通道,连接), 错误 抓住 _ER1:_ER2 -> close_pipe(Channel2,Connection2), 错误 结尾。

将容错添加到具体的 erlang 应用程序中,利用上面的一些代码,您将拥有一个非常紧凑的集群。 RABBITMQ 的可用性非常高,特别是如果您在了解自己在做什么的同时使用它。大多数问题都是由网络引起的。如果您的应用程序失去与 Rabbitmq 服务器的连接,您的应用程序会做什么?冗余在这里出现。您可以拥有各种RABBITMQ 服务器,所有集群节点都知道这些服务器,并且您在所有 RABBITMQ 服务器上安装了所有相同的交换和队列,这样如果一台服务器丢失,其他服务器可以用于继续服务。

【讨论】:

  • 我认为您已经使用 rabbitmq 来链接您的多系统。但我将它用于我的系统和第三方程序。另一个程序不能被信任,所以我必须限制它的行为,以免浪费服务器的资源。
猜你喜欢
  • 2021-03-01
  • 1970-01-01
  • 1970-01-01
  • 2014-07-31
  • 2018-06-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多