servicestack - 使用 ServiceStack 广播 rabbitMq 消息

Question

有没有办法制作方法

myMessageService.CreateMessageQueueClient().Publish(myMessage);

向所有接收者广播消息？

Answer 1

问题是，RegisterHandler<T>内部使用类型 T构建它监听的队列名称。因此，您唯一的机会就是通过对多个队列使用自定义扇出交换来使用以下解决方案偏离轨道:

var fanoutExchangeName = string.Concat(QueueNames.Exchange,
                                       ".",
                                       ExchangeType.Fanout);

在您系统的某些点，您必须确保使用以下代码进行交换:

var rabbitMqServer = new RabbitMqServer();
var messageProducer = (RabbitMqProducer) rabbitMqServer.CreateMessageProducer();
var channel = messageProducer.Channel; // you can use any logic to acquire a channel here - this is just a demo
channel.ExchangeDeclare(fanoutExchangeName,
                        ExchangeType.Fanout,
                        durable: true,
                        autoDelete: false,
                        arguments: null);

现在我们可以向这个扇出发布消息:

var message = new Message<T>(yourInstance);
messageProducer.Publish(QueueNames<T>.In,    // routing key
                        message,             // message
                        fanoutExchangeName); // exchange

现在消息已发布到我们的交换器，但我们需要将队列绑定(bind)到消费组件中的交换器，我们这样做:

var rabbitMqServer = new RabbitMqServer();
var messageQueueClient = (RabbitMqQueueClient) rabbitMqServer.CreateMessageQueueClient();
var channel = messageQueueClient.Channel; // you just need to get the channel

var queueName = messageQueueClient.GetTempQueueName();
channel.QueueBind(queueName,          // queue
                  fanoutExchangeName, // exchange
                  QueueName<T>.In);   // routing key

队列在最后一个(也是唯一一个)消费者断开连接后自动删除，并且不会在 RabbitMq 重启后继续存在。

hacky 部分现在是听...

var consumer = new RabbitMqBasicConsumer(channel);
channel.BasicConsume(queueName,
                     false,
                     consumer);

Task.Run(() =>
            {
                while (true)
                {
                    BasicGetResult basicGetResult;
                    try
                    {
                        basicGetResult = consumer.Queue.Dequeue();
                    }
                    catch (EndOfStreamException endOfStreamException)
                    {
                        // this is ok
                        return;
                    }
                    catch (OperationInterruptedException operationInterruptedException)
                    {
                        // this is ok
                        return;
                    }
                    catch (Exception ex)
                    {
                        throw;
                    }
                    var message = basicGetResult.ToMessage<T>();
                    // TODO processing
                }
            });

虽然此解决方案不提供任何自动重新连接、过滤器或其他功能。

基本演练是 available here .

编辑:我刚刚想到的一件事:您可以使用 ServiceStack.Messaging.MessageHandler<T>轻松提供回复和重试的实例。