dotnetty 新的篇章 - 开源 - Powered by Discuz! Archiver

栋淮發表於 2025-6-20 12:59:00

dotnetty 新的篇章- 开源

<h2>一、前言</h2>
      因为微服务引擎依赖于dotnetty组件，很多协议都是针对于dotnetty 进行扩展，然后对于老版本https://github.com/azure/dotnetty 停止更新后，本人下载源码进行修改更新，并且大家要求独立仓库进行开源，所以今天整理了代码开源至https://github.com/microsurging/DotNetty,  也希望大家一起贡献代码，让dotnetty 生态更强大。
HttpFlv:http://demo.kayakiot.cn:281/httpflv.html  （黑衣人）
 HttpFlv:http://demo.kayakiot.cn:281/httpflv1.html  （大红包）
HttpFlv:http://demo.kayakiot.cn:281/httpflv2.html  （鹿鼎记）
rtmp:rtmp://demo.kayakiot.cn:76/live1/livestream2   （黑衣人）
rtmp:rtmp://demo.kayakiot.cn:76/live1/livestream3   （大红包）
rtmp:rtmp://demo.kayakiot.cn:76/live1/livestream4（鹿鼎记）
注：测试服务器带宽只有8MB, httpflv  缓冲做的没有rtmp好，然后httpflv卡就多刷新几次
  凯亚 (Kayak) 是什么?
       凯亚(Kayak)是基于.NET8.0软件环境下的surging微服务引擎进行开发的, 平台包含了微服务和物联网平台。支持异步和响应式编程开发，功能包含了物模型,设备,产品,网络组件的统一管理和微服务平台下的注册中心，服务路由，模块，中间服务等管理。还有多协议适配(TCP,MQTT,UDP,CoAP,HTTP,Grpc,websocket,rtmp,httpflv,webservice,等),通过灵活多样的配置适配能够接入不同厂家不同协议等设备。并且通过设备告警,消息通知,数据可视化等功能。能够让你能快速建立起微服务物联网平台系统。
     凯亚物联网平台：http://demo.kayakiot.cn:3100（用户名：fanly  密码：123456）
    链路跟踪Skywalking V8:http://117.72.121.2:8080/
      surging 微服务引擎开源地址：https://github.com/fanliang11/surging（后面surging 会移动到microsurging进行维护）
<h2>二、ValueTask扩展支持</h2>
IValueTaskPromise：
<div class="cnblogs_code">
<pre> public interface IValueTaskPromise: IPromise
{
 bool IsVoid { get; }

 bool IsCompleted { get; }

 bool IsSuccess { get; }

 bool IsFaulted { get; }

 bool IsCanceled { get; }

 bool TryComplete();

 void Complete();

 bool TrySetException(Exception exception);

 bool TrySetException(IEnumerable<Exception> exceptions);

 void SetException(Exception exception);

 void SetException(IEnumerable<Exception> exceptions);

 bool TrySetCanceled();

 void SetCanceled();

 bool SetUncancellable();

 IPromise Unvoid();
 }</pre>
</div>
DefaultValueTaskPromise：
<div class="cnblogs_code">
<pre> public class DefaultValueTaskPromise: IValueTaskPromise
{
 private readonly CancellationToken _token;
#if NET
 private readonly TaskCompletionSource _tcs;
#else
 private readonly ManualResetValueTaskSource<object> _tcs;
#endif

 private int v_uncancellable = SharedConstants.False;

 public DefaultValueTaskPromise()
 {
 _token = CancellationToken.None;
#if NET
 _tcs = new TaskCompletionSource();
#else
 _tcs = new ManualResetValueTaskSource<object>();
#endif
 }

 public DefaultValueTaskPromise(object state)
 {
#if NET
 _tcs = new TaskCompletionSource(state);
#else
 _tcs = new ManualResetValueTaskSource<object>(state);
#endif
 }

 public DefaultValueTaskPromise(CancellationToken cancellationToken)
 {
 _token= cancellationToken;
 }

 public ValueTask ValueTask
 {

 get => _tcs.AwaitVoid(_token);
 }

 public bool IsVoid => false;

 public bool IsSuccess => ValueTask.IsCompletedSuccessfully;

 public bool IsCompleted => ValueTask.IsCompleted;

 public bool IsFaulted => ValueTask.IsFaulted;

 public bool IsCanceled => ValueTask.IsCanceled;

 publicTaskTask => ValueTask.AsTask();

 public virtual bool TryComplete()
 {
#if NET
 return _tcs.TrySetResult();
#else
 return _tcs.SetResult(0);
#endif
 }

 public virtual void Complete()
 {
#if NET
 _tcs.SetResult();
#else
 _tcs.SetResult(0);
#endif
 }
 public virtual void SetCanceled()
 {
 if (SharedConstants.False < (uint)Volatile.Read(ref v_uncancellable)) { return; }
 _tcs.SetCanceled();
 }

 public virtual void SetException(Exception exception)
 {
 if (exception is AggregateException aggregateException)
 {
 SetException(aggregateException.InnerExceptions);
 return;
 }
 _tcs.SetException(exception);
 }

 public virtual void SetException(IEnumerable<Exception> exceptions)
 {
 _tcs.SetException(exceptions.FirstOrDefault());
 }

 public virtual bool TrySetCanceled()
 {
 if (SharedConstants.False < (uint)Volatile.Read(ref v_uncancellable)) { return false; }
 _tcs.SetCanceled();
 return true;
 }

 public virtual bool TrySetException(Exception exception)
 {
 if (exception is AggregateException aggregateException)
 {
 return TrySetException(aggregateException.InnerExceptions);
 }
 _tcs.SetException(exception);
 return true;
 }

 public virtual bool TrySetException(IEnumerable<Exception> exceptions)
 {
 _tcs.SetException(exceptions.FirstOrDefault());
 return true;
 }

 public bool SetUncancellable()
 {
 if (SharedConstants.False >= (uint)Interlocked.CompareExchange(ref v_uncancellable, SharedConstants.True, SharedConstants.False))
 {
 return true;
 }
 return !IsCompleted;
 }

 public override string ToString() => "TaskCompletionSource";

 public IPromise Unvoid() => this;

}</pre>
</div>
ManualResetValueTaskSource：
<div class="cnblogs_code">
<pre> internal interface IStrongBox<T>
{
ref T Value { get; }

bool RunContinuationsAsynchronously { get; set; }
}

public enum ContinuationOptions
{
None,

ForceDefaultTaskScheduler
}

public class ManualResetValueTaskSource<T> : IStrongBox<ManualResetValueTaskSourceLogic<T>>, IValueTaskSource<T>, IValueTaskSource
{
private ManualResetValueTaskSourceLogic<T> _logic;
private readonly Action _cancellationCallback;

public ManualResetValueTaskSource(ContinuationOptions options = ContinuationOptions.None)
{
 _logic = new ManualResetValueTaskSourceLogic<T>(this, options,null);
 _cancellationCallback = SetCanceled;
}

public ManualResetValueTaskSource(object state, ContinuationOptions options = ContinuationOptions.None)
{
 _logic = new ManualResetValueTaskSourceLogic<T>(this, options,state);
 _cancellationCallback = SetCanceled;
}

public short Version => _logic.Version;

public bool SetResult(T result)
{
 lock (_cancellationCallback)
 {
 if (_logic.Completed)
 {
 return false;
 }

 _logic.SetResult(result);
 return true;
 }
}

public void SetException(Exception error)
{
 if (Monitor.TryEnter(_cancellationCallback))
 {
 if (_logic.Completed)
 {
 Monitor.Exit(_cancellationCallback);
 return;
 }

 _logic.SetException(error);
 Monitor.Exit(_cancellationCallback);
 }
}

public void SetCanceled() => SetException(new TaskCanceledException());

public T GetResult(short token) => _logic.GetResult(token);

void IValueTaskSource.GetResult(short token) => _logic.GetResult(token);

public ValueTaskSourceStatus GetStatus(short token) => _logic.GetStatus(token);

public bool RunContinuationsAsynchronously { get; set; } = true;

public void OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags) => _logic.OnCompleted(continuation, state, token, flags);

ref ManualResetValueTaskSourceLogic<T> IStrongBox<ManualResetValueTaskSourceLogic<T>>.Value => ref _logic;

public ValueTask<T> AwaitValue(CancellationToken cancellation)
{
 CancellationTokenRegistration? registration = cancellation == CancellationToken.None
 ? (CancellationTokenRegistration?)null
 : cancellation.Register(_cancellationCallback);
 return _logic.AwaitValue(this, registration);
}

public ValueTask AwaitVoid(CancellationToken cancellation)
{
 CancellationTokenRegistration? registration = cancellation == CancellationToken.None
 ? (CancellationTokenRegistration?)null
 : cancellation.Register(_cancellationCallback);
 return _logic.AwaitVoid(this, registration);
}

public void Reset() => _logic.Reset();
}

internal struct ManualResetValueTaskSourceLogic<TResult>
{
private static readonly Action<object> s_sentinel = s => throw new InvalidOperationException();

private readonly IStrongBox<ManualResetValueTaskSourceLogic<TResult>> _parent;
private readonly ContinuationOptions _options;
private Action<object> _continuation;
private object _continuationState;
private object _capturedContext;
private ExecutionContext _executionContext;
private bool _completed;
private TResult _result;
private ExceptionDispatchInfo _error;
private CancellationTokenRegistration? _registration;

public ManualResetValueTaskSourceLogic(IStrongBox<ManualResetValueTaskSourceLogic<TResult>> parent, ContinuationOptions options,object state)
{
 _parent = parent ?? throw new ArgumentNullException(nameof(parent));
 _options = options;
 _continuation = null;
 _continuationState = null;
 _capturedContext = null;
 _executionContext = null;
 _completed = state != null;
 _result =state==null? default(TResult): (TResult)state;
 _error = null;
 Version = 0;
 _registration = null;
}

public short Version { get; private set; }

public bool Completed => _completed;

private void ValidateToken(short token)
{
 if (token != Version)
 {
 throw new InvalidOperationException();
 }
}

public ValueTaskSourceStatus GetStatus(short token)
{
 // ValidateToken(token);

 return
 !_completed ? ValueTaskSourceStatus.Pending :
 _error == null ? ValueTaskSourceStatus.Succeeded :
 _error.SourceException is OperationCanceledException ? ValueTaskSourceStatus.Canceled :
 ValueTaskSourceStatus.Faulted;
}

public TResult GetResult(short token)
{
 // ValidateToken(token);

 if (!_completed)
 {
 return _result;
 }

 TResult result = _result;
 ExceptionDispatchInfo error = _error;
 Reset();

 error?.Throw();
 return result;
}

public void Reset()
{
 Version++;

 _registration?.Dispose();

 _completed = false;
 _continuation = null;
 _continuationState = null;
 _result = default(TResult);
 _error = null;
 _executionContext = null;
 _capturedContext = null;
 _registration = null;
}

public void OnCompleted(Action<object> continuation, object state, short token, ValueTaskSourceOnCompletedFlags flags)
{
 if (continuation == null)
 {
 throw new ArgumentNullException(nameof(continuation));
 }

 ValidateToken(token);

 if ((flags & ValueTaskSourceOnCompletedFlags.FlowExecutionContext) != 0)
 {
 _executionContext = ExecutionContext.Capture();
 }

 if ((flags & ValueTaskSourceOnCompletedFlags.UseSchedulingContext) != 0)
 {
 SynchronizationContext sc = SynchronizationContext.Current;
 if (sc != null && sc.GetType() != typeof(SynchronizationContext))
 {
 _capturedContext = sc;
 }
 else
 {
 TaskScheduler ts = TaskScheduler.Current;
 if (ts != TaskScheduler.Default)
 {
 _capturedContext = ts;
 }
 }
 }

 _continuationState = state;
 if (Interlocked.CompareExchange(ref _continuation, continuation, null) != null)
 {
 _executionContext = null;

 object cc = _capturedContext;
 _capturedContext = null;

 switch (cc)
 {
 case null:
 Task.Factory.StartNew(continuation, state, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);
 break;

 case SynchronizationContext sc:
 sc.Post(s =>
 {
 var tuple = (Tuple<Action<object>, object>)s;
 tuple.Item1(tuple.Item2);
 }, Tuple.Create(continuation, state));
 break;

 case TaskScheduler ts:
 Task.Factory.StartNew(continuation, state, CancellationToken.None, TaskCreationOptions.DenyChildAttach, ts);
 break;
 }
 }
}

public void SetResult(TResult result)
{
 _result = result;
 SignalCompletion();
}

public void SetException(Exception error)
{
 _error = ExceptionDispatchInfo.Capture(error);
 SignalCompletion();
}

private void SignalCompletion()
{
 if (_completed)
 {
 throw new InvalidOperationException("Double completion of completion source is prohibited");
 }

 _completed = true;

 if (Interlocked.CompareExchange(ref _continuation, s_sentinel, null) != null)
 {
 if (_executionContext != null)
 {
 ExecutionContext.Run(
 _executionContext,
 s => ((IStrongBox<ManualResetValueTaskSourceLogic<TResult>>)s).Value.InvokeContinuation(),
 _parent ?? throw new InvalidOperationException());
 }
 else
 {
 InvokeContinuation();
 }
 }
}

private void InvokeContinuation()
{
 object cc = _capturedContext;
 _capturedContext = null;

 if (_options == ContinuationOptions.ForceDefaultTaskScheduler)
 {
 cc = TaskScheduler.Default;
 }

 switch (cc)
 {
 case null:
 if (_parent.RunContinuationsAsynchronously)
 {
 var c = _continuation;
 if (_executionContext != null)
 {
 ThreadPool.QueueUserWorkItem(s => c(s), _continuationState);
 }
 else
 {
 ThreadPool.UnsafeQueueUserWorkItem(s => c(s),_continuationState);
 }
 }
 else
 {
 _continuation(_continuationState);
 }
 break;

 case SynchronizationContext sc:
 sc.Post(s =>
 {
 ref ManualResetValueTaskSourceLogic<TResult> logicRef = ref ((IStrongBox<ManualResetValueTaskSourceLogic<TResult>>)s).Value;
 logicRef._continuation(logicRef._continuationState);
 }, _parent ?? throw new InvalidOperationException());
 break;

 case TaskScheduler ts:
 Task.Factory.StartNew(_continuation, _continuationState, CancellationToken.None, TaskCreationOptions.DenyChildAttach, ts);
 break;
 }
}

public ValueTask<T> AwaitValue<T>(IValueTaskSource<T> source, CancellationTokenRegistration? registration)
{
 _registration = registration;
 return new ValueTask<T>(source, Version);
}

public ValueTask AwaitVoid(IValueTaskSource source, CancellationTokenRegistration? registration)
{
 _registration = registration;
 return new ValueTask(source, Version);
}
}</pre>
</div>
然后把DefaultPromise 替换成DefaultValueTaskPromise，如下图所示
<img src="https://img2024.cnblogs.com/blog/192878/202506/192878-20250620124957180-2094341490.png">
<h2>三、扩展支持rtmp编解码</h2>
如下图所示：
<img src="https://img2024.cnblogs.com/blog/192878/202506/192878-20250620125136757-1795393625.png">
<h2>四、demo展示</h2>
开启了三个通道进行推流，cpu,内存都比较稳定
<img src="https://img2024.cnblogs.com/blog/192878/202506/192878-20250620125507980-1788282642.png">
 在凯亚物联网平台你也可以创建rtmp组件
<img src="https://img2024.cnblogs.com/blog/192878/202506/192878-20250620125753939-1623893860.png">
 
视频中心
<img src="https://img2024.cnblogs.com/blog/192878/202506/192878-20250620125921979-1655464165.png">
  
来源：https://www.cnblogs.com/fanliang11/p/18938169

頁: [1]

琼殿技术论坛's Archiver

dotnetty 新的篇章- 开源