9、rabbitmq

public class RabbitMQContext
{
// 获取或设置交换机名称。
public string ExchangeName { get; set; }
// 获取或设置 RoutingKey 。
public string RoutingKey { get; set; }
// 获取或设置队列名称。
public string QueueName { get; set; }
// 获取或设置 .
public string ExchangeType { get; set; }
// 获取或设置发布消息的属性。
public IDictionary Headers { get; set; }
// 获取或设置定义 Exchange 时的参数。
public IDictionary ExchangeArgs { get; set; }
public bool IsDelayMessage { get; set; }
}
public class RabbitMQConsumerContext : RabbitMQContext
{
}
public class RabbitMQProducerContext : RabbitMQContext
{
public RabbitMQMessage Body { get; set; }
}
public class RabbitMQMessage
{
private object _message;
[JsonProperty(PropertyName = "message")]
public object Message
{
get
{
return this._message;
}
set
{
this._message = value;
this.MessageType = value == null
? string.Empty
: value.GetType().FullName;
}
}
[JsonProperty("messageType")]
public string MessageType { get; set; }
}

public interface IRabbitMQPublisher
{
// 发布消息到消息队列。
bool Publish(RabbitMQProducerContext context);
}

public class RabbitMQPublisher : IRabbitMQPublisher
{
private readonly IConfiguration _configProvider;
private readonly RabbitMQOptions _options;
public RabbitMQPublisher(IConfiguration configProvider, IOptions options)
{
this._cOnfigProvider= configProvider;
this._optiOns= options.Value;
}
public bool Publish(RabbitMQProducerContext context)
{
if (string.IsNullOrWhiteSpace(context.ExchangeName))
{
context.ExchangeName = _configProvider["RabbitMQ:Default:Exchange"];
}
var factory = new ConnectionFactory
{
HostName = this._options.Host,
Port = this._options.Port,
UserName = this._options.UserName,
Password = this._options.Password
};
using (var cOnnection= factory.CreateConnection())
using (var channel = connection.CreateModel())
{
var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(context.Body));
channel.ExchangeDeclare(context.ExchangeName, context.ExchangeType, durable: true, autoDelete: false, arguments: context.ExchangeArgs);
if (!string.IsNullOrWhiteSpace(context.QueueName))
{
channel.QueueDeclare(context.QueueName, durable: true, exclusive: false, autoDelete: false, arguments: context.ExchangeArgs);
channel.QueueBind(context.QueueName, context.ExchangeName, context.RoutingKey);
}
IBasicProperties basicProperties = null;
if (context.Headers != null && context.Headers.Any())
{
basicProperties = channel.CreateBasicProperties();
basicProperties.Headers = context.Headers;
}
channel.BasicPublish(context.ExchangeName, context.RoutingKey, basicProperties: basicProperties, body: body);
}
return true;
}
}

[HttpGet]
public JsonResult RabbitMqPublish([FromQuery] string paramStr)
{
var cOntext= new RabbitMQProducerContext
{
Body = new RabbitMQMessage
{
Message = JsonConvert.SerializeObject(new { paramStr })
},
QueueName = "new_born",
RoutingKey = "new_born",
ExchangeType = ExchangeType.Topic
};
var optiOnValue= this._configuration.GetSection("RabbitMQ").Get();
var optiOns= (IOptions)Options.Create(optionValue);
var publisher = new RabbitMQPublisher(this._configuration, options);
publisher.Publish(context);
return Json(new { code = 1, message = "发布成功" });
}

public interface IMessageQueueConsumer
{
// 订阅消息。
void Subscribe();
// 取消订阅消息。
void Unsubscribe();
// 通知生产者此消息已被消费
void BasicAck(ulong deliveryTag, bool multiple);
}

///

public class MessageQueueHostedService : IHostedService
{
private readonly IServiceProvider _serviceProvider;
private readonly RabbitMQOptions _options;
public MessageQueueHostedService(IServiceProvider serviceProvider, IOptions options)
{
this._serviceProvider = serviceProvider;
this._optiOns= options.Value;
}
public Task StartAsync(CancellationToken cancellationToken)
{
if (!this._options.Enable) return Task.CompletedTask;
var cOnsumers= this._serviceProvider.GetServices();
if (consumers.Any())
{
foreach (var consumer in consumers)
{
consumer.Subscribe();
}
}
return Task.CompletedTask;
}
public Task StopAsync(CancellationToken cancellationToken)
{
if (!this._options.Enable) return Task.CompletedTask;
return Task.CompletedTask;
}
}

public class NewBornConsumer : RabbitMQConsumer
{
private IServiceProvider _serviceProvider;
public NewBornConsumer(IServiceProvider serviceProvider) : base(serviceProvider)
{
_serviceProvider = serviceProvider;
}
protected override void Initialize()
{
this.COntext= new RabbitMQConsumerContext
{
ExchangeType = ExchangeType.Topic,
ExchangeName = "new_born",
QueueName = "new_born",
RoutingKey = "new_born",
};
}
protected async override Task OnConsumerReceivedAsync(object sender, DeliverEventArgs e)
{
await Task.CompletedTask;
var message = Encoding.UTF8.GetString(e.Body.ToArray());
var data = JToken.Parse(message)["message"].ToString();
var paramStr = JToken.Parse(data)["paramStr"];
Logger.LogInformation($"我被消费拉 ~~~~参数是：{paramStr}");
this.BasicAck(e.DeliveryTag, false);
}
}