使用EventBus + Redis发布订阅模式提升业务执行性能(下)

前言

上一篇博客上已经实现了使用eventbus对具体事件行为的分发处理，某种程度上也算是基于事件驱动思想编程了。但是如上篇博客结尾处一样，我们源码的执行效率依然达不到心里预期。在下单流程里我们明显可以将部分行为进行异步处理，提升下单操作的执行效率。

redis基础命令

redis有两种方式可支持我们实现mq功能，1、使用列表(list)相关命令特性；2、使用publish、subscribe命令特性；
这里我是采取列表相关命令实现。

使用列表(list)相关命令的特性实现

• 压入数据(发布消息)
  使用列表(list)的 lpush rpush 命令可以从列表左边和右边压入数据;

lpush
将一个或多个值插入到列表头部(此处可以将列表想象成一个从左到右的链表数据结构，lpush就是将指定的值插入最左侧！)

 如下命令，将多个元素压入list1的头部(最左侧)

lpush list1 测试1 测试2

执行结果如下:

 

 

 

上面是写入多个元素，我们也可以写入单个元素

lpush list1 测试3

　　

 

 

 

需要留意，每次执行完lpush后，redis会返回当前列表的长度。

rpush
在指定列表的尾部(相当于一个链表的最右侧)添加单个或多个元素

如下命令，还是在list1上添加多个元素，并查看执行后的list1元素信息

rpush list 测试4 测试5

　　

 

 

 

同理，rpush也可直接写入单个元素，和lpush一样。

 

• 拉取数据(消费数据)
  这里的拉取数据不单单是读取list内的元素，而是将元素从列表中取出来

blpop
移出并获取列表的第一个元素(从左至右)， 如果列表没有元素会阻塞当前线程，直到等待超时或发现可弹出元素为止。

如下命令，从list1这个列表获取从左至右第一个元素，在100秒内如果获取则结束阻塞，否则阻塞到100秒之后。

blpop list1 100

　　

执行结果如下：

 

 

需要留意的是blpop命令如果拉取到数据则会返回两行数据，1行为列表的key名称，1行为获取到的元素值。如果直到阻塞结束都没有获取到元素值则直接返回命令执行超时。如下图:

 

 

brpop
移出并获取列表的最后一个元素(从左至右)， 如果列表没有元素会阻塞当前线程直到等待超时或发现可弹出元素为止。该命令与blpop除了获取的元素位置不同，其他特性全部一致。

lpop
移出并获取列表的第一个元素(从左至右)，如获取到元素则返回元素信息，没有元素则立即返回null。

如下命令:

lpop list1

　　

 

 

rpop
移出并获取列表的最后一个元素(从左至右)，如获取到元素则返回元素信息，没有元素则立即返回null。该命令与lpop除了获取的元素位置不同其他特性全部一致；

rpoplpush
移除列表的最后一个元素(最右侧的元素)，并将该元素添加到另一个列表并返回。该命令如获取到元素则返回元素信息，否则返回错误信息。

可以通过rpoplpush这个命令的特性对mq内一致性要求较高的业务进行处理，在从列表获取元素成功后将该元素添加到一个备份列表，在业务处理完毕后再从备份列表将该元素删除。
执行下面命令测试下:

rpoplpush list1 listback

　　

 

 

brpoplpush
从列表中弹出一个值，将弹出的元素插入到另外一个列表中并返回它； 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。

该命令其实就是在brpop的基础上将lpush的功能加上了，依旧也保留了指定超时时间内未获取到元素则阻塞线程。
执行下面命令测试下:

brpoplpush list1 listback 10

　　

执行结果如下:

 

 

完善代码

基于上面redis的相关命令，我们再完善下上篇博客的代码。这里我们需要新增一个控制台启动项，将它作为消费服务，原来的控制台即订单保存的控制台作为消息发布的服务。
下单代码更改为下面的样子:

        /// <summary>
        /// 异步方式触发订单相关事件
        /// </summary>
        public static void asyneventhandle()
        {
            guid userid = guid.newguid();
            stopwatch stopwatch = new stopwatch();
            stopwatch.start();

            var order = new ordermodel()
            {
                createtime = datetime.now,
                id = guid.newguid(),
                money = (decimal)300.00,
                number = 1,
                productname = "鲜花一束",
                userid = userid
            };
            console.writeline($"模拟存储订单【采取redis做消息队列的异步方式】");
            thread.sleep(1000);

            fullredis fullredis = new fullredis("127.0.0.1:6379", "", 1);
            //这里尝试过使用redis 的订阅发布模式,在执行发布命令时候发现值但凡出现空格或者"符号则会异常...         
            fullredis.lpush("orders", new ordermodel[] { order });

            stopwatch.stop();
            console.writeline($"下单总耗时:{stopwatch.elapsedmilliseconds}毫秒");
            console.readline();
        }

　　

可以看到，我们已经将事件总线相关代码给移除了，上面代码除了向redis的队列(list)里写入元素外就只是对订单进行了持久化动作，所以看代码就知道执行效率的提升了。
接下来，看消费服务的代码。

        static void main(string[] args)
        {
            xtrace.useconsole();
            console.writeline("进入redis消息订阅者模式订单消息推送订阅者客户端!");

            eventbus eventbus = new eventbus();
            eventbus.eventregister(typeof(ordercreateeventnotifyhandle), typeof(ordercreateeventdata));
            eventbus.eventregister(typeof(ordercreateeventstocklockhandle), typeof(ordercreateeventdata));

            fullredis fullredis = new fullredis("127.0.0.1:6379", "", 1);
            fullredis.log = xtrace.log;
            fullredis.timeout = 30000;
            ordermodel order = null;
            while (order == null)
            { 
                order = fullredis.blpop<ordermodel>("orders", 20);
                if (order != null)
                {
                    console.writeline($"得到订单信息:{jsonconvert.serializeobject(order)}");
                    //执行相关事件
                    eventbus.trigger(new ordercreateeventdata()
                    {
                        order = order,
                    });
                    //再次设置为null方便循环读取
                    order = null;
                }
                  
            }
            console.readline();
        }

　　

消费服务首先从redis里通过blpop从orders列表中获取元素，再触发事件总线，执行订单保存相关业务处理。

最终看下执行效率如何？
消息发布的执行效率(订单保存)

 

消息消费

 

可以看到目前消息发布的执行效率下单总耗时间为1170毫秒，我们再改为同步的测试下结果：

 

可以看到，同步执行的结果是3035毫秒。

小结

两种方式相差了将近2000毫秒~ 而且后续如果再继续扩展订单存储相关处理的话同步执行的响应时间会更加拉长，而采取redis mq的方式配合事件总线我们可以将整个业务拆分为独立的应用，采取分布式的方式提高响应效率，同时事件总线的加入方便我们后续业务的扩展。

消息发布端将订单信息写入到列表后如果消息消费者在拉取到数据后业务执行过程中代码出现异常导致无法满足业务的完整性如何处理

答:可以使用上述redis命令中的rpoplpush或brpoplpush在拉取元素后写入到一个备份的列表中，在我们的逻辑代码执行完毕后在将备份列表中的该元素值移除。

上述代码已发布到github，有需要的自行下载。

地址为:https://github.com/qq897878763/orderredissample