平泉建设局网站,体育设施 网站模版,河南省建设监理协会网站人才十,怎么购买域名和服务器聆听自己的声音如果自己学不动了#xff0c;或者感觉没有动力的时候#xff0c;看看书#xff0c;听听音乐#xff0c;跑跑步#xff0c;休息两天#xff0c;重新出发#xff0c;偷懒虽好#xff0c;可不要贪杯。话说上回书我们说到了#xff0c;Redis的使用修改《【B… 聆听自己的声音如果自己学不动了或者感觉没有动力的时候看看书听听音乐跑跑步休息两天重新出发偷懒虽好可不要贪杯。话说上回书我们说到了Redis的使用修改《【BCVP更新】StackExchange.Redis的异步开发方式》通过异步的时候基本上会解决StackExRedis组件使用过程中可能在并发的时候遇到的问题而且该组件也是微软官方推荐的参考微软微服务框架eShopOnContainers如果一定要抬杠说不好用其实是没必要的。那今天我们继续往下说简单说下如何基于Redis实现消息队列。目前在市面上比较主流的消息队列中间件主要有Kafka、ActiveMQ、RabbitMQ、RocketMQ等这几种。当然常见的还是基于RabbitMQ来实现的Redis份额稍微小了一点但是因为Redis的仓储、缓存等多个方面的好处使得Redis也是很火。1什么是消息队列这个其实我今天不打算重点讲因为我详细每个人能看这篇文章肯定都知道消息队列的相关内容但是为了不那么突兀我就从网上粘贴几块基本概念了解一二基本概念消息队列英语Message queue是一种进程间通信或同一进程的不同线程间的通信方式软件的贮列用来处理一系列的输入通常是来自用户。消息队列提供了异步的通信协议每一个贮列中的纪录包含详细说明的数据包含发生的时间输入设备的种类以及特定的输入参数也就是说消息的发送者和接收者不需要同时与消息队列交互。消息会保存在队列中直到接收者取回它。最终可以实现解耦的目的。下面通过一个简单的架构模型来解释Producer消息生产者负责产生和发送消息到Broker。Broker消息处理中心。负责消息存储、确认、重试等一般其中会包含多个Queue。Consumer消息消费者负责从 Broker 中获取消息并进行相应处理。有哪些优缺点从上边的定义中我们可以看出来优点主要是三块异步、流量削峰与流控、解耦。这三个优点在高并发等三高场景还是很有必要的甚至说是十分必要的。典型的广播模式一个消息可以发布到多个消费者消息即时发送消息不用等待消费者读取消费者会自动接收到信道发布的消息比如我们某宝下订单或某6抢车票那都是放到队列里缓冲的要是都用服务端等待可能早就崩了当然实际上比这个复杂的多。而且通过订阅发布的模式异步执行这样就会大大缓解时间压力。但是随之而来的弊端也是有的比如为了异步就是接收者必须轮询消息队列才能收到最近的消息。然后还有就是不能达到实时性说白了就是用空间换时间从而降低瓶颈。消息一旦发布不能接收。换句话就是发布时若客户端不在线则消息丢失不能寻回。不能保证每个消费者接收的时间是一致的。若消费者客户端出现消息积压到一定程度会被强制断开导致消息意外丢失。五种常见模式简单模式Hello World功能一个生产者P发送消息到队列Q,一个消费者C接收工作队列模式Work Queue功能一个生产者多个消费者每个消费者获取到的消息唯一多个消费者只有一个队列发布/订阅模式Publish/Subscribe功能一个生产者发送的消息会被多个消费者获取。一个生产者、一个交换机、多个队列、多个消费者路由模式Routing说明生产者发送消息到交换机并且要指定路由key消费者将队列绑定到交换机时需要指定路由key通配符主题模式Topic说明生产者P发送消息到交换机Xtypetopic交换机根据绑定队列的routing key的值进行通配符匹配更多具体的内容呢自己感兴趣多去搜索下吧肯定还是有很多其他问题的我这里就不铺开了讲了下边咱们就说说如何在Blog.Core里添加队列吧。2订阅发布相关配置案例案例有很多自己可以根据情况自定义。那既然要讲东西肯定不能随便放一个算法肯定是需要一个小demo一个应用场景这样更有助于初学者去理解之前考虑了很多一直没有想好在BlogCore里边使用什么案例场景来说一说消息队列最后实在是没办法只能说日志了万事不决就说日志好像软件开发都是这么举例的。这里说一下假设我们自定义了一个日志记录的方法就是在txt里写数据其实我现在也是这么用的平时肯定会一边查一边写如果并发高一下肯定就会出现死锁或者异常的出现那我们就可以把写日志放到消息队列里缓冲一下然后在写一个订阅者专门来“盯着”队列一有消息传过来就写到日志文件里这样就能很好的实现相应的目的。如果不缓冲下有时候日志可能高达几万条瞬间爆炸。那说了这个小场景接下来就简单的模拟一下吧。1、定义消息队列操作类与接口既然要发布和订阅消息肯定就需要有相应的操作方法在上一篇文章中我新建了一个RedisBasketRepository.cs的操作类那我们还继续在这个类文件里写吧注意这个实现类和接口已经注册到服务容器了如果你第一次操作可以参考文章开头上篇文章内容 /// summary/// 根据key获取RedisValue/// /summary/// typeparam nameT/typeparam/// param nameredisKey/param/// returns/returnspublic async TaskRedisValue[] ListRangeAsync(string redisKey){return await _database.ListRangeAsync(redisKey);}/// summary/// 在列表头部插入值。如果键不存在先创建再插入值/// /summary/// param nameredisKey/param/// param nameredisValue/param/// returns/returnspublic async Tasklong ListLeftPushAsync(string redisKey, string redisValue, int db -1){return await _database.ListLeftPushAsync(redisKey, redisValue);}/// summary/// 在列表尾部插入值。如果键不存在先创建再插入值/// /summary/// param nameredisKey/param/// param nameredisValue/param/// returns/returnspublic async Tasklong ListRightPushAsync(string redisKey, string redisValue, int db -1){return await _database.ListRightPushAsync(redisKey, redisValue);}/// summary/// 移除并返回存储在该键列表的第一个元素 反序列化/// /summary/// param nameredisKey/param/// returns/returnspublic async TaskT ListLeftPopAsyncT(string redisKey, int db -1) where T : class{return JsonConvert.DeserializeObjectT(await _database.ListLeftPopAsync(redisKey));}
我这里只是简单的Copy出来几个做例子总的一共有12个当然你也可以自定义增加或删除某些不必要的核心的可以看出来都是根据redisKey来操作的TaskRedisValue[] ListRangeAsync(string redisKey);Tasklong ListLeftPushAsync(string redisKey, string redisValue, int db -1);Tasklong ListRightPushAsync(string redisKey, string redisValue, int db -1);Tasklong ListRightPushAsync(string redisKey, IEnumerablestring redisValue, int db -1);TaskT ListLeftPopAsyncT(string redisKey, int db -1) where T : class;TaskT ListRightPopAsyncT(string redisKey, int db -1) where T : class;Taskstring ListLeftPopAsync(string redisKey, int db -1);Taskstring ListRightPopAsync(string redisKey, int db -1);Tasklong ListLengthAsync(string redisKey, int db -1);TaskIEnumerablestring ListRangeAsync(string redisKey, int db -1);TaskIEnumerablestring ListRangeAsync(string redisKey, int start, int stop, int db -1);Tasklong ListDelRangeAsync(string redisKey, string redisValue, long type 0, int db -1);Task ListClearAsync(string redisKey, int db -1);
2、如何发布消息与接收消息上边定义好了相应的操作方法以后就很简单了我们来发布一条消息来试试 [HttpGet][AllowAnonymous]public async Task RedisMq(){var msg 这里是一条日志;await _redisBasketRepository.ListLeftPushAsync(RedisMqKey.Loging, msg);}
就是这么简单构造函数注入以后直接调用相应的方法就把消息msg推送到了队列里了这里的redisKey我用了常量定义具体可操作Blog.Core源代码。现在是发布消息特别简单只需要一行接口那如何去获取呢在上边的获取方法中我们定义的是TaskRedisValue[] ListRangeAsync(string redisKey);
这个方法也是可以的只不过我们需要对其进行转换毕竟存的msg是字符串string类型的但是这里的返回类型的RedisValue[]所以需要劈里啪啦转化一下。但是这里有一个问题就是如何去定时获取呢也就是如何设计一个订阅者进行消费消息呢这需要思考下当然比较简单的就是while(true){}可能平时就是这么使用的不过还是不是那么爽快可以写一个组件来处理简单快捷正好有一个大佬已经封装好了我们可以直接拿来用如果你有什么问题可以给他提issue。3、InitQ组件来订阅消息在nuget中可以直接安装组个组件PackageReference IncludeInitQ Version1.0.0.4 /
他的开源地址是https://github.com/wmowm/Initq使用方法很简单可以参考他的README里的介绍1、先添加服务 /// summary
/// Redis 消息队列 启动服务
/// /summary
public static class RedisInitMqSetup
{public static void AddRedisInitMqSetup(this IServiceCollection services){if (services null) throw new ArgumentNullException(nameof(services));services.AddInitQ(m {//时间间隔m.SuspendTime 5000;//redis服务器地址m.ConnectionString 127.0.0.1:6379;//对应的订阅者类需要new一个实例对象当然你也可以传参比如日志对象m.ListSubscribe new ListIRedisSubscribe() { new RedisSubscribe()};//显示日志m.ShowLog false;});}
}
2、定义订阅者 public class RedisSubscribe : IRedisSubscribe{[Subscribe(RedisMqKey.Loging)]private async Task SubRedisLoging(string msg){Console.WriteLine($队列{RedisMqKey.Loging} 消费到/接受到 消息:{msg});await Task.CompletedTask;}}整体很简单继承接口然后添加上特性这个特性里的参数就是我们消息发布的时候的那个key然后方法的参数就是对应的消息msg是不是很简单。当然这里你可以传递一个日志的对象实例这样就把日志信息分流到了队列里然后队列走到这个订阅者里由这里进行缓冲然后把日志填充到日志文件从而达到减峰的目的。最终的效果可以看看好啦今天的redis消息队列已经说完了还是很简单的其中重点还是那五种模式要自己好好了解下然后整体过程自己把握把握至于RabbitMQ这个以后再说吧。END扫码关注老张的哲学更多精彩等着你
