.net core Redis消息队列中间件【InitQ】

标签：out   字符   消息发布   pen   load   ref   消息队列   targe   scope   

前言

这是一篇拖更很久的博客,不知不觉InitQ在nuget下载量已经过15K了,奈何胸无点墨也不晓得怎么写(懒),随便在github上挂了个md,现在好好唠唠如何在redis里使用队列

队列缓存分布式 异步调优堆配置 ------(来自某位不知名码友)

诞生背景

redis在项目中使用的越来越频繁,通常我们是用来做缓存,使用较多的就是String,Hash这两种类型,以及分布式锁,redis的List类型,就可以用于消息队列,使用起来更加简单,且速度更快,非常适合子服务内部之间的消息流转,创造灵感来自于杨老板的CAP(地址:https://www.cnblogs.com/tibos/p/11858095.html),采用注解的方式消费队列,让业务逻辑更加的清晰,方便维护

安装环境

• .net core版本：2.1
• redis版本：3.0以上

特点

1.通过注解的方式,订阅队列
2.可以设置消费消息的频次
3.支持消息广播
4.支持延迟队列

使用介绍

• 1.获取initQ包

  方案A. install-package InitQ
  方案B. nuget包管理工具搜索 InitQ

• 2.添加中间件(该中间件依赖 StackExchange.Redis)

  services.AddInitQ(m=> 
  {
      m.SuspendTime = 1000;
      m.IntervalTime = 1000; 
      m.ConnectionString = "127.0.0.1,connectTimeout=15000,syncTimeout=5000,password=123456";
      m.ListSubscribe = new List() { typeof(RedisSubscribeA), typeof(RedisSubscribeB) };
      m.ShowLog = false;
  });
  

• 3.配置说明

  public class InitQOptions
  {
      /// 
  
      /// redis连接字符串
      /// 
      public string ConnectionString { get; set; }
  
      /// 
  
      /// 没消息时挂起时长(毫秒)
      /// 
      public int SuspendTime { get; set; }
  
      /// 
  
      /// 每次消费消息间隔时间(毫秒)
      /// 
      public int IntervalTime { get; set; }
  
      /// 
  
      /// 是否显示日志
      /// 
      public bool ShowLog { get; set; }
  
      /// 
  
      /// 需要注入的类型
      /// 
      public IList ListSubscribe { get; set; }
  
      public InitQOptions()
      {
          ConnectionString = "";
          IntervalTime = 0;
          SuspendTime = 1000;
          ShowLog = false;
      }
  }
  

消息发布/订阅

消息的发布/订阅是最基础的功能,这里做了几个优化

1. 采用的是长轮询模式,可以控制消息消费的频次,以及轮询空消息的间隔,避免资源浪费
2. 支持多个类订阅消息,可以很方便的根据业务进行分类,前提是这些类 必须注册
3. 支持多线程消费消息(在执行耗时任务的时候,非常有用)

示例如下(Thread.Sleep):

    public class RedisSubscribeA: IRedisSubscribe
    {
        [Subscribe("tibos_test_1")]
        private async Task SubRedisTest(string msg)
        {
            Console.WriteLine($"A类--->当前时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")} 订阅者A消费消息:{msg}");
            Thread.Sleep(3000); //使用堵塞线程模式,同步延时
            Console.WriteLine($"A类

    public class RedisSubscribeA: IRedisSubscribe
    {
        [Subscribe("tibos_test_1")]
        private async Task SubRedisTest(string msg)
        {
            Console.WriteLine($"A类--->当前时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")} 订阅者A消费消息:{msg}");
            Thread.Sleep(3000); //使用堵塞线程模式,同步延时
            Console.WriteLine($"A类当前时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")} 订阅者A消费消息:{msg}");
            Thread.Sleep(3000); //使用堵塞线程模式,同步延时
            Console.WriteLine($"A类

示例如下(Task.Delay):

    [Subscribe("tibos_test_1")]
    private async Task SubRedisTest(string msg)
    {
        Console.WriteLine($"A类--->当前时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")} 订阅者A消费消息:{msg}");
        await Task.Delay(3000); //使用非堵塞线程模式,异步延时
        Console.WriteLine($"A类

根据业务情况,合理的选择堵塞模式

• 1.订阅发布者

    using (var scope = _provider.GetRequiredService().CreateScope())
    {
        //redis对象
        var _redis = scope.ServiceProvider.GetService();
        //循环向 tibos_test_1 队列发送消息
        for (int i = 0; i 

• 2.定义消费者类 RedisSubscribeA

  public class RedisSubscribeA: IRedisSubscribe
  {
      [Subscribe("tibos_test_1")]
      private async Task SubRedisTest(string msg)
      {
          Console.WriteLine($"A类--->订阅者A消息消息:{msg}");
      }
  
      [Subscribe("tibos_test_1")]
      private async Task SubRedisTest1(string msg)
      {
          Console.WriteLine($"A类--->订阅者A1消息消息:{msg}");
      }
  
      [Subscribe("tibos_test_1")]
      private async Task SubRedisTest2(string msg)
      {
          Console.WriteLine($"A类--->订阅者A2消息消息:{msg}");
      }
  
      [Subscribe("tibos_test_1")]
      private async Task SubRedisTest3(string msg)
      {
          Console.WriteLine($"A类--->订阅者A3消息消息:{msg}");
      }
  }
  

• 3.定义消费者类 RedisSubscribeB

  public class RedisSubscribeB : IRedisSubscribe
  {
      /// 
  
      /// 测试
      /// 
      /// 
      /// 
      [Subscribe("tibos_test_1")]
      private async Task SubRedisTest(string msg)
      {
          Console.WriteLine($"B类--->订阅者B消费消息:{msg}");
      }
  }
  

消息广播/订阅

消息广播是StackExchange.Redis已经封装好的,我们只用起个线程监听即可,只要监听了这个key的线程,都会收到消息

• 1.订阅消息通道,订阅者需要在程序初始化的时候启动一个线程侦听通道,这里使用HostedService来实现,并注册到容器

    public class ChannelSubscribeA : IHostedService, IDisposable
    {
        private readonly IServiceProvider _provider;
        private readonly ILogger _logger;
  
        public ChannelSubscribeA(ILogger logger, IServiceProvider provider)
        {
            _logger = logger;
            _provider = provider;
        }
        public void Dispose()
        {
            _logger.LogInformation("退出");
        }
  
        public Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("程序启动");
            Task.Run(async () =>
            {
                using (var scope = _provider.GetRequiredService().CreateScope())
                {
                    //redis对象
                    var _redis = scope.ServiceProvider.GetService();
                    await _redis.SubscribeAsync("test_channel", new Action((channel, message) =>
                    {
                        Console.WriteLine("test_channel" + " 订阅服务A收到消息：" + message);
                    }));
  
                }
            });
            return Task.CompletedTask;
        }
  
        public Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("结束");
            return Task.CompletedTask;
        }
    }
  

    public class ChannelSubscribeB : IHostedService, IDisposable
    {
        private readonly IServiceProvider _provider;
        private readonly ILogger _logger;
  
        public ChannelSubscribeB(ILogger logger, IServiceProvider provider)
        {
            _logger = logger;
            _provider = provider;
        }
        public void Dispose()
        {
            _logger.LogInformation("退出");
        }
  
        public Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("程序启动");
            Task.Run(async () =>
            {
                using (var scope = _provider.GetRequiredService().CreateScope())
                {
                    //redis对象
                    var _redis = scope.ServiceProvider.GetService();
                    await _redis.SubscribeAsync("test_channel", new Action((channel, message) =>
                    {
                        Console.WriteLine("test_channel" + " 订阅服务B收到消息：" + message);
                    }));
  
                }
            });
            return Task.CompletedTask;
        }
  
        public Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("结束");
            return Task.CompletedTask;
        }
    }
  

• 2.将HostedService类注入到容器

    services.AddHostedService();
    services.AddHostedService();
  

• 3.广播消息

    using (var scope = _provider.GetRequiredService().CreateScope())
    {
        //redis对象
        var _redis = scope.ServiceProvider.GetService();
        for (int i = 0; i 

延迟消息

延迟消息非常适用处理一些定时任务的场景,如订单15分钟未付款,自动取消, xxx天后,自动续费...... 这里使用zset+redis锁来实现,这里的操作方式,跟发布/订阅非常类似
写入延迟消息:SortedSetAddAsync
注解使用:SubscribeDelay

• 1.定义发布者

    Task.Run(async () =>
    {
  
        using (var scope = _provider.GetRequiredService().CreateScope())
        {
            //redis对象
            var _redis = scope.ServiceProvider.GetService();
  
            for (int i = 0; i 

• 2.定义消费者

    //延迟队列
    [SubscribeDelay("test_0625")]
    private async Task SubRedisTest1(string msg)
    {
        Console.WriteLine($"A类--->当前时间:{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")} 订阅者延迟队列消息开始--->{msg}");
        //模拟任务执行耗时
        await Task.Delay(TimeSpan.FromSeconds(3));
        Console.WriteLine($"A类--->{msg} 结束

版本

• V1.0 更新时间:2019-12-30

版本库：

• Git获取：https://github.com/wmowm/InitQ

作者：提伯斯

.net core Redis消息队列中间件【InitQ】

标签：out   字符   消息发布   pen   load   ref   消息队列   targe   scope   

原文地址：https://www.cnblogs.com/tibos/p/14944832.html