您可以创建一个托管服务并在其中使用您的消息。此托管服务始终处于活动状态，并且可以接收所有消息。
public class ProductTopicConsumerService : ConsumerBase, IHostedService
{
public ProductTopicConsumerService(ConnectionFactory connectionFactory)
: base(connectionFactory, ExchangeTypes.topic)
{
try
{
Consume<ProductCreatedIntegrationEvent>();
}
catch (Exception ex)
{
Console.WriteLine($"Error => {ex?.Message}");
}
}
protected override string Exchange => "ProductExchangeTopic5";
protected override string Queue => "Product.Updated5";
protected override string AppId => "ProductCreatedConsumer";
protected override string QueueAndExchangeRoutingKey => "Product.*";
public virtual Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
public virtual Task StopAsync(CancellationToken cancellationToken)
{
Dispose();
return Task.CompletedTask;
}
}
和 ConsumerBase 可能是这样的：
public abstract class ConsumerBase : RabbitMqClientBase
{
public ConsumerBase(ConnectionFactory connectionFactory, ExchangeTypes exchangeType)
: base(connectionFactory, exchangeType,false)
{
}
protected void Consume<TMessage>()
{
var consumer = new AsyncEventingBasicConsumer(Channel);
consumer.Received += OnEventReceived<TMessage>;
Channel.BasicConsume(queue: RefinedQueueName, autoAck: false, consumer: consumer);
}
private Task OnEventReceived<TMessage>(object sender, BasicDeliverEventArgs @event)
{
try
{
var body = Encoding.UTF8.GetString(@event.Body.ToArray());
var message = JsonConvert.DeserializeObject<TMessage>(body);
ReceiveAction(message);
Channel.BasicAck(@event.DeliveryTag, false);
return Task.CompletedTask;
}
catch (Exception ex)
{
Console.WriteLine($"Error occurred in OnEventReceived. Message: {ex?.Message}");
Channel.BasicNack(@event.DeliveryTag, false, true);
throw;
}
finally
{
}
}
protected virtual void ReceiveAction<TMessage>(TMessage message)
{
}
}
最后，rabbitMQBase 可能是这样的：
public abstract class RabbitMqClientBase : IDisposable
{
protected const string VirtualHost = "MQ";
protected abstract string Exchange { get; }
protected abstract string Queue { get; }
protected abstract string AppId { get; }
protected abstract string QueueAndExchangeRoutingKey { get; }
protected IModel Channel { get; private set; }
private IConnection _connection;
private readonly ConnectionFactory _connectionFactory;
private readonly ExchangeTypes _exchangeType;
private readonly bool _isPublisher;
protected RabbitMqClientBase(ConnectionFactory connectionFactory, ExchangeTypes exchangeType, bool isPublisher)
{
_connectionFactory = connectionFactory;
this._exchangeType = exchangeType;
this._isPublisher = isPublisher;
ConnectToRabbitMq();
}
protected internal string RefinedExchangeName => $"{VirtualHost}.{Exchange}";
protected internal string RefinedQueueName => $"{VirtualHost}.{Queue}";
protected internal string RefinedRoutingKey => $"{VirtualHost}.{QueueAndExchangeRoutingKey}";
private void ConnectToRabbitMq()
{
if (_connection == null || _connection.IsOpen == false)
{
_connection = _connectionFactory.CreateConnection();
}
if (Channel == null || Channel.IsOpen == false)
{
Channel = _connection.CreateModel();
Channel.ExchangeDeclare(exchange: RefinedExchangeName, type: _exchangeType.ToString(), durable: true, autoDelete: false);
if (!_isPublisher)
{
Channel.QueueDeclare(queue: RefinedQueueName, durable: true, exclusive: false, autoDelete: false);
Channel.QueueBind(queue: RefinedQueueName, exchange: RefinedExchangeName, routingKey: RefinedRoutingKey);
}
}
}
public void Dispose()
{
Channel?.Close();
Channel?.Dispose();
Channel = null;
_connection?.Close();
_connection?.Dispose();
_connection = null;
}
}
这是 RabbitMQ 的 Publisher|Prodcuer 类：
public interface IRabbitMqProducer<in T>
{
PublishResult Publish(T @event);
}
public abstract class ProducerBase<T> : RabbitMqClientBase, IRabbitMqProducer<T>
{
protected ProducerBase(ConnectionFactory connectionFactory, ExchangeTypes exchangeType)
: base(connectionFactory, exchangeType, true) { }
public PublishResult Publish(T @event)
{
var result = new PublishResult();
try
{
var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(@event));
var properties = Channel.CreateBasicProperties();
properties.AppId = AppId;
properties.ContentType = "application/json";
properties.DeliveryMode = 2; //persist mode
properties.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds());
Channel.BasicPublish(exchange: RefinedExchangeName, routingKey: RefinedRoutingKey, basicProperties: properties, body: body);
result.SetSuccess();
}
catch (Exception ex)
{
result.SetError(ex);
}
return result;
}
}
为了发布你的 publisher 应该是这样的：
public class ProductTopicProducerService : ProducerBase<ProductCreatedIntegrationEvent>
{
public ProductTopicProducerService(ConnectionFactory connectionFactory)
: base(connectionFactory,ExchangeTypes.topic)
{
}
protected override string Exchange => "ProductExchangeTopic5";
protected override string Queue => "";
protected override string AppId => "ProductCreatedConsumer";
protected override string QueueAndExchangeRoutingKey => "Product.*";
}
对于发布新消息，请遵循以下代码：
var result = _producer.Publish(@event);
if (!result.IsSuccess)
{
Console.WriteLine($"Error => {result.Description}");
}
外包您可以简化您的消费者。但这是我为 rabbitMQBase 创建基类的经验，现在发布者和消费者都可以使用它。