查看原文
其他

.NET Core 使用 Channel 消息队列

DotNet 2024-04-12

推荐关注↓

背景
最近做一个项目,连接了很多设备,需要保存设备的心跳数据,刚开始的做法是直接接收到设备的数据之后进行心跳数据的保存,但是随着设备多了起来,然后设备的使用时长不断的加大,对数据库的压力也比较大,所以想着优化一下。
方案调研
1、使用第三方中间件

常见的使用redis,或者mq,只需要不断的向中间件发送数据即可,redis使用队列,如果是mq直接发送消息即可,使用起来简单方便,但是要引入这些中间件,目前的架构里面没有,需要自己去起服务,维护。

2、使用channel
System.Threading.Channels 是.NET Core 3.0 后推出的新的集合类型, 具有异步API,高性能,线程安全等特点,它可以用来做消息队列,进行数据的生产和消费, 公开的 Writer 和 Reader api对应消息的生产者和消费者,也让Channel更加的简洁和易用,与Rabbit MQ 等其他队列不同的是,Channel 是进程内的队列
目前就介绍来看非常完美,不需要添加第三方中间件,直接添加现有的模块即可。
代码实现
选择了使用channel来做优化。拿到设备数据之后直接把消息丢入到channel,然后后台使用定时任务或者自己实现hostservice去不断的消费数据。
生产者代码
public async Task ProduceHeartBeat(string message)
{
    await channel.Writer.WriteAsync(message);
}

不断的向里面写入数据即可.

消费者代码

/// <summary>
/// timespan时间内消费多少数据
/// </summary>
/// <param name="count"></param>
/// <param name="timeSpan"></param>
/// <returns></returns>
public async Task<List<string>> ConsumeHeartBeatAsync(int count,TimeSpan timeSpan)
{
     var result = new List<string>(count);
     CancellationTokenSource cts = new CancellationTokenSource();
     var cancellationToken = cts.Token;
     cts.CancelAfter(timeSpan);
     int rcount = 0;
     while ( !cancellationToken.IsCancellationRequested && rcount<count)
     {
         //await Task.Delay(2000);
         if (channel.Reader.TryRead(out var number))
         {
             Console.WriteLine(number);
             result.Add(number);
             rcount++;
         }
         else
         {
             break;
         } 
     }  
    return result;
}

里面加入了一个cancellationToken,进行消费的时长限制。在此时长内消费多少条数据,超时直接结束。

这就是基本的代码

后台定时消费数据

public class HeartBeatService : BackgroundService
{
     private readonly HeartBeatsChannel heartBeatsChannel;

     public HeartBeatService(HeartBeatsChannel heartBeatsChannel)
     {
         this.heartBeatsChannel = heartBeatsChannel;
     }

     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
         try
         {

             Task.Factory.StartNew(() =>
             {
                 while (!stoppingToken.IsCancellationRequested)
                 {
                     //阻塞的队列使得一直在同一个线程运行
                     Process(15,heartBeatsChannel).Wait();
                 }

             }, TaskCreationOptions.LongRunning);

             Console.WriteLine("主线程 现在运行的线程id为:" + Thread.CurrentThread.ManagedThreadId);

             }
         catch (Exception ex)
         {
             Console.WriteLine(ex.ToString());
         }
     }
     /// <summary>
     /// 消费数据
     /// </summary>
     /// <param name="count">一次消费数量</param>
     /// <param name="heartBeatsChannel"></param>
     /// <returns></returns>
     private async Task Process(int count ,HeartBeatsChannel heartBeatsChannel)
     {
         Console.WriteLine("子线程_现在运行的线程id为:" + Thread.CurrentThread.ManagedThreadId);
         //每次消费三十个
         if (heartBeatsChannel.IsHasContent)
         {
             //int count = 15;
             //进行消费
             await heartBeatsChannel.ConsumeHeartBeatAsync(count, TimeSpan.FromSeconds(3));
         }           
         await Task.Delay(3000);
     }
}

使用的是BackgroundServic,直接实现要处理的业务逻辑就好了。

在这里使用的是TaskCreationOptions.LongRunning,新开一个线程去处理心跳数据。


总结


以上就是主要的实现全过程,完整的代码在github

https://github.com/lackguozi/LearnChannelWebApi

实际上完全可以不用后台去定时消费数据,channel有很多api可以去处理,比如WaitToReadAsync(),但是这里没有使用,主要是不想持续的占数据库资源?

总结的话学习了channel的用法,底层似乎使用了deque?只稍微看了下源码,但是看到了许多的lock,这个是必不可少的。还是巨硬轮子造的好

转自:果小天

链接:cnblogs.com/guoxiaotian/p/17506536.html

- EOF -

推荐阅读  点击标题可跳转
C# FFmpeg 音视频开发总结
C# 实现 Linux 视频聊天、远程桌面(源码).NET Core 使用 HttpClient 的正确方式

看完本文有收获?请转发分享给更多人

推荐关注「DotNet」,提升.Net技能 

点赞和在看就是最大的支持❤️

继续滑动看下一个
向上滑动看下一个

您可能也对以下帖子感兴趣

文章有问题?点此查看未经处理的缓存