引言
.Net TPL Dataflow是一个进程内数据流管道，应对高并发、低延迟的要求非常有效， 但在实际Docker部署的过程中， 有一个问题一直无法回避：
单体程序部署的瞬间（服务不可用）会有少量流量无法处理；
更糟糕的情况下，迭代部署的这个版本有问题，上线后无法工作， 导致更多流量没有处理。
背负神圣使命（巨大压力）的程序猿心生一计，为何不将单体程序改成分布式：
增加服务ReceiverApp，ReceiverApp只接受数据，WebApp只处理数据。
知识储备
消息队列和订阅发布作为老生常谈的两个知识点被反复提及，按照JMS的规范， 官方称为点对点(point to point， queue)和发布/订阅（publish/subscribe，channel）
点对点
生产者发送消息到Message Queue中，然后消费者从队列中取出消息并消费。
队列会保留消息，直到他们被消费或超时；
① MQ支持多消费者，但每个消息只能被一个消费者处理
② 发送者和消费者在时间上没有依赖性，当发送者发送消息之后，不管消费者有没有在运行（甚至不管有没有消费者），都不会影响到消息被发送到队列
③ 一般消费者在消费之后需要向队列应答成功
如果希望发送的每个消息都被成功处理，你应该使用p2p模型
发布/订阅
消息生产者将消息发布到Channel，在此之前已有多个消费者订阅该通道。
和点对点方式不同，发布到特定通道的消息会被通道订阅者实时接收。
通道没有队列机制，发布的消息只能被当前收听的订阅者接收到
① 每个消息可以有多个订阅者
② 发布者和消费者有时间上依赖性：某通道的订阅者，必须先创建该通道订阅，才能收到消息
发布消息至通道，不关注订阅者是谁；订阅者可收听自己感兴趣的多个通道（类似于topic），也不关注发布者是谁。
③ 故如果没有订阅者，发布的消息将得不到处理；
头脑风暴
Redis内置的List数据结构能形成轻量级消息队列的效果；Redis原生支持发布/订阅 模型
如上分析， Pub/Sub模型在订阅者宕机的时候，发布的消息得不到处理，故此模型不能用于强业务的数据接收和处理。
本次采用的消息队列模型：
- 解耦业务：新建ReceiverApp作为生产者，专注于接收并发送到队列；原有的WebApp作为消费者专注数据处理。
- 起到削峰填谷的作用，若缩放出多个WebApp消费者容器，还能形成负载均衡的效果。
需要关注Redis操作List结构的两个命令( 左进右出，右进左出同理)：
LPUSH & RPOP/BRPOP
Brpop中的B 表示"Block",是一个rpop命令的阻塞版本：若指定List没有新元素，在给定时间内，该命令会阻塞当前redis客户端连接，直到超时返回nil
AspNetCore编程实践
本次使用AspNetCore 完成RedisMQ的实践，引入Redis国产第三方开源库CSRedisCore
生产者ReceiverApp
生产者使用LPush命令向Redis List数据结构写入消息。
------------------截取自Startup.cs------------------------- public void ConfigureServices(IServiceCollection services) { // Redis客户端要定义成单例， 不然在大流量并发收数的时候， 会造成redis client来不及释放。另一方面也确认api控制器不是单例模式， var csredis = new CSRedisClient(Configuration.GetConnectionString("redis")+",name=receiver"); RedisHelper.Initialization(csredis); services.AddSingleton(csredis); services.AddMvc(); } ------------------截取自数据接收Controller-------------------[Route("batch")] [HttpPost] public async Task BatchPutEqidAndProfileIds([FromBody]List<EqidPair> eqidPairs){ if (!ModelState.IsValid) throw new ArgumentException("Http Body Payload Error."); var redisKey = $"{DateTime.Now.ToString("yyyyMMdd")}"; eqidPairs = await EqidExtractor.EqidExtractAsync(eqidPairs); if (eqidPairs != null && eqidPairs.Any()) RedisHelper.LPush(redisKey, eqidPairs.ToArray()); await Task.CompletedTask; }
消费者WebApp
根据以上RedisMQ思路，事件消费方式是拉取pull，故需要轮询Redis List数据结构，这里使用AspNetCore内置的BackgroundService后台服务类后台轮询消费：
关注后台Job中的循环接收方法。
public class BackgroundJob : BackgroundService { private readonly IEqidPairHandler _eqidPairHandler; private readonly CSRedisClient[] _cSRedisClients; private readonly I Configuration _conf; private readonly ILogger _logger; public BackgroundJob(IEqidPairHandler eqidPairHandler, CSRedisClient[] csRedisClients,IConfiguration conf,ILoggerFactory loggerFactory) { _eqidPairHandler = eqidPairHandler; _cSRedisClients = csRedisClients; _conf = conf; _logger = loggerFactory.CreateLogger(nameof(BackgroundJob)); } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _logger.LogInformation("Service starting"); if (_cSRedisClients[0] == null) { _cSRedisClients[0] = new CSRedisClient(_conf.GetConnectionString("redis") + ",defaultDatabase=" + 0); } RedisHelper.Initialization(_cSRedisClients[0]); while (!stoppingToken.IsCancellationRequested) { var key = $"eqidpair:{DateTime.Now.ToString("yyyyMMdd")}"; var eqidpair = RedisHelper.BRPop(5, key); if (eqidpair != null) await_ eqidPairHandler.AcceptEqidParamAsync(JsonConvert.DeserializeObject<EqidPair>(eqidpair)); // 强烈建议无论如何休眠一段时间，防止突发大流量导致WebApp进程CPU满载，自行根据场景设置合理休眠时间 await Task.Delay(10, stoppingToken); } _logger.LogInformation("Service stopping"); } }
迭代验证
使用docker-compose单机部署Nginx,ReceiverApp,WebApp容器。
docker-compose up指令默认只会重建[Service配置或Image变更]的容器。
If there are existing containers for a service, and the service’s configuration or image was changed after the container’s creation,
docker-compose uppicks up the changes by stopping and recreating the containers (preserving mounted volumes). To prevent Compose from picking up changes, use the--no-recreateflag.
做一次迭代验证，更新docke-compose.yml文件WebApp服务的镜像版本，
docker-compose up；
下图显示仅 数据处理容器 WebApp被Recreate：
Nice,分布式改造完成，效果很明显，现在可以放心安全的迭代核心WebApp数据处理程序。
