82 lines
2.9 KiB
C#
82 lines
2.9 KiB
C#
using System.Diagnostics;
|
|
using BSI.Stack.Rabbit;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Options;
|
|
using RabbitMQ.Client;
|
|
using Volo.Abp.DependencyInjection;
|
|
|
|
namespace SJZY.Track.Application.Rabbit;
|
|
|
|
/// <summary>
|
|
/// 支持发布确认模式
|
|
/// </summary>
|
|
public class RabbitClientBusExt : RabbitClientBus, ISingletonDependency
|
|
{
|
|
public RabbitClientBusExt(IServiceProvider sp, ILogger<RabbitClient> logger)
|
|
: base(sp, logger, sp.GetRequiredService<IOptionsMonitor<RabbitOptions>>().CurrentValue)
|
|
{
|
|
}
|
|
|
|
public void PublishByConfirmSelect(
|
|
string exchange,
|
|
string routingKey,
|
|
byte[] data,
|
|
Action<IBasicProperties>? options = null
|
|
)
|
|
{
|
|
using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey);
|
|
activity?.SetTag("kind", "publish");
|
|
var key = exchange + ":" + routingKey;
|
|
Lazy<IModel> orAdd = Channels.GetOrAdd(key, (Func<string, Lazy<IModel>>) (_ => new Lazy<IModel>(CreateModel)));
|
|
var basicProperties1 = orAdd.Value.CreateBasicProperties();
|
|
|
|
// 配置属性
|
|
if (options != null)
|
|
options(basicProperties1);
|
|
var model = orAdd.Value;
|
|
|
|
// 启用发布确认
|
|
model.ConfirmSelect();
|
|
var readOnlyMemory = (ReadOnlyMemory<byte>) data;
|
|
model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory);
|
|
model.WaitForConfirmsOrDie();
|
|
}
|
|
|
|
public void PublishByConfirmSelect(
|
|
string exchange,
|
|
string routingKey,
|
|
IEnumerable<byte[]> datas,
|
|
Action<IBasicProperties>? options = null
|
|
)
|
|
{
|
|
using var activity = RabbitInstrumentation.ActivitySource.StartActivity("/" + exchange + "/" + routingKey);
|
|
activity?.SetTag("kind", "publish");
|
|
var key = exchange + ":" + routingKey;
|
|
Lazy<IModel> orAdd = Channels.GetOrAdd(key, (Func<string, Lazy<IModel>>) (_ => new Lazy<IModel>(CreateModel)));
|
|
var basicProperties1 = orAdd.Value.CreateBasicProperties();
|
|
|
|
// 配置属性
|
|
if (options != null)
|
|
options(basicProperties1);
|
|
var model = orAdd.Value;
|
|
|
|
// 启用发布确认
|
|
model.ConfirmSelect();
|
|
foreach (var data in datas)
|
|
{
|
|
var readOnlyMemory = (ReadOnlyMemory<byte>) data;
|
|
model.BasicPublish(exchange, routingKey, basicProperties1, readOnlyMemory);
|
|
}
|
|
model.WaitForConfirmsOrDie();
|
|
}
|
|
|
|
private IConnection GetConnection()
|
|
{
|
|
var key = Connections.OrderBy((Func<KeyValuePair<IConnection, int>, int>) (q => q.Value)).Select((Func<KeyValuePair<IConnection, int>, IConnection>) (q => q.Key)).FirstOrDefault();
|
|
Connections.AddOrUpdate(key, (Func<IConnection, int>) (_ => 1), (Func<IConnection, int, int>) ((_, v) => v + 1));
|
|
return key;
|
|
}
|
|
|
|
private IModel CreateModel() => GetConnection().CreateModel();
|
|
}
|