disruptor 用户封装自己的消费者,把消费者注入到消费者容器,消费者容器实现自动创建 缓存队列,生产者;
文中用到的 disruptor C#移植源代码
https://github.com/bingyang001/disruptor-net-3.3.0-alpha
作者博客 http://www.cnblogs.com/liguo/p/3296166.html
消费者容器:
/// <summary>
/// 消费者管理器
/// </summary>
/// <typeparam name="TProduct">产品</typeparam>
public class Workers<TProduct> where TProduct : Producer<TProduct>, new()
{
private readonly WorkerPool<TProduct> _workerPool;
public Workers(List<IWorkHandler<TProduct>> handers, IWaitStrategy waitStrategy = null, int bufferSize = 1024*64)
{
if (handers == null || handers.Count == 0)
throw new ArgumentNullException("消费事件处理数组为空!");
if (handers.Count == 1)
_ringBuffer = RingBuffer<TProduct>.CreateSingleProducer(() => new TProduct(), bufferSize,
waitStrategy ?? new YieldingWaitStrategy());
else
{
_ringBuffer = RingBuffer<TProduct>.CreateMultiProducer(() => new TProduct(), bufferSize,
waitStrategy ?? new YieldingWaitStrategy());
}
_workerPool = new WorkerPool<TProduct>(_ringBuffer
, _ringBuffer.NewBarrier()
, new FatalExceptionHandler()
, handers.ToArray());
_ringBuffer.AddGatingSequences(_workerPool.getWorkerSequences());
}
public void Start()
{
_workerPool.start(TaskScheduler.Default);
}
public Producer<TProduct> CreateOneProducer()
{
return new Producer<TProduct>(this._ringBuffer);
}
public void DrainAndHalt()
{
_workerPool.drainAndHalt();
}
private readonly RingBuffer<TProduct> _ringBuffer;
}
生产者(产品): 所有的产品都应该继承自生产者
/// <summary>
/// 生产者对象
/// </summary>
/// <typeparam name="TProduct">产品类型</typeparam>
public class Producer<TProduct> where TProduct:Producer<TProduct>
{
long _sequence;
private RingBuffer<TProduct> _ringBuffer;
public Producer()
{
}
public Producer(RingBuffer<TProduct> ringBuffer )
{
_ringBuffer = ringBuffer;
}
/// <summary>
/// 获取可修改的产品
/// </summary>
/// <returns></returns>
public Producer<TProduct> Enqueue()
{
long sequence = _ringBuffer.Next();
Producer<TProduct> producer = _ringBuffer[sequence];
producer._sequence = sequence;
if (producer._ringBuffer == null)
producer._ringBuffer = _ringBuffer;
return producer;
}
/// <summary>
/// 提交产品修改
/// </summary>
public void Commit()
{
_ringBuffer.Publish(_sequence);
}
}
--------------------------------------------------------
以上就实现了,测试代码
先创建 产品对象:
/// <summary>
/// 产品/继承生产者
/// </summary>
public class Product : Producer<Product>
{
//产品包含的属下随便定义,无要求,只需要继承自生产者就行了
public long Value { get; set; }
public string Guid { get; set; }
}
创建消费者对象
/// <summary>
/// 消费处理对象
/// </summary>
public class WorkHandler : IWorkHandler<Product>
{
public void OnEvent(Product @event)
{
//Test是测试对象数据准确(数据重复或者丢失数据)
Test.UpdateCacheByOut(@event.Guid);
//收到产品,在这里写处理代码
}
}
测试代码:
可创建1个或者多个的生产者对象,消费者处理对象;不一定太多,多不一定快; 建议生产者创建一个就行了,多线程操作一个生产者对象; 消费者对象可以根据实际情况创建多少个;
//创建2个消费者,2个生产者, 2个消费者表示,框架会有2个线程去处理消费产品 Workers<Product> workers = new Workers<Product>(
new List<IWorkHandler<Product>>() {new WorkHandler(), new WorkHandler()});
Producer<Product> producerWorkers = workers.CreateOneProducer();
Producer<Product> producerWorkers1 = workers.CreateOneProducer(); //开始消费 workers.Start();
产品生产:
可以在任何引用生产者的地方,把产品放进队列中. 这里 放入队列的方法和平时不太一样. 这里采用的是,从队列里面拿去一个位置,然后把产品放进去; 具体的做法 ,找生产者,获取一个产品对象,然后修改产品属性,最后提交修改.
var obj = producer.Enqueue();
//修改产品属性
obj.Commit();
以上是关键代码:
完整的测试类 : 包含测试数据正确性, 性能,在不校验正确性的时候,每秒ops 1千万左右.
class Test
{
public static long PrePkgInCount = 0;
public static long PrePkgOutCount = 0;
public static long PkgInCount = 0;
public static long PkgOutCount = 0;
static ConcurrentDictionary<string, string> InCache = new ConcurrentDictionary<string, string>();
static ConcurrentDictionary<string, string> OutCache = new ConcurrentDictionary<string, string>();
private static long Seconds;
static void Main(string[] args)
{
Workers<Product> workers = new Workers<Product>(
new List<IWorkHandler<Product>>() {new WorkHandler(), new WorkHandler()});
Producer<Product> producerWorkers = workers.CreateOneProducer();
Producer<Product> producerWorkers1 = workers.CreateOneProducer();
workers.Start();
Task.Run(delegate
{
while (true)
{
Thread.Sleep(1000);
Seconds++;
long intemp = PkgInCount;
long outemp = PkgOutCount;
Console.WriteLine(
$"In ops={intemp - PrePkgInCount},out ops={outemp - PrePkgOutCount},inCacheCount={InCache.Count},OutCacheCount={OutCache.Count},RunningTime={Seconds}");
PrePkgInCount = intemp;
PrePkgOutCount = outemp;
}
});
Task.Run(delegate { Run(producerWorkers); });
Task.Run(delegate { Run(producerWorkers); });
Task.Run(delegate { Run(producerWorkers1); });
Console.Read();
}
public static void Run(Producer<Product> producer)
{
for (int i = 0; i < int.MaxValue; i++)
{
var obj = producer.Enqueue();
CheckRelease(obj as Product);
obj.Commit();
}
}
public static void CheckRelease(Product publisher)
{
Interlocked.Increment(ref PkgInCount);
return; //不检查正确性
publisher.Guid = Guid.NewGuid().ToString();
InCache.TryAdd(publisher.Guid, string.Empty);
}
public static void UpdateCacheByOut(string guid)
{
Interlocked.Increment(ref Test.PkgOutCount);
if (guid != null)
if (InCache.ContainsKey(guid))
{
string str;
InCache.TryRemove(guid, out str);
}
else
{
OutCache.TryAdd(guid, string.Empty);
}
}
/// <summary>
/// 产品/继承生产者
/// </summary>
public class Product : Producer<Product>
{
//产品包含的属下随便定义,无要求,只需要继承自生产者就行了
public long Value { get; set; }
public string Guid { get; set; }
}
/// <summary>
/// 消费处理对象
/// </summary>
public class WorkHandler : IWorkHandler<Product>
{
public void OnEvent(Product @event)
{
Test.UpdateCacheByOut(@event.Guid);
//收到产品,在这里写处理代码
}
}
}
时间： 2024-10-23 02:26:43