C#实现异步消息队列

简介: 原文:C#实现异步消息队列拿到新书《.net框架设计》,到手之后迅速读了好多,虽然这本书不像很多教程一样从头到尾系统的讲明一些知识,但是从项目实战角度告诉我们如何使用我们的知识,从这本书中提炼了一篇,正好符合我前几篇的“数据驱动框架”设计的问题; 消息队列 消息队列(英语:Message queue)是一种进程间通信或同一进程的不同线程间的通信方式,软件的贮列用来处理一系列的输入,通常是来自使用者。
原文: C#实现异步消息队列

拿到新书《.net框架设计》,到手之后迅速读了好多,虽然这本书不像很多教程一样从头到尾系统的讲明一些知识,但是从项目实战角度告诉我们如何使用我们的知识,从这本书中提炼了一篇,正好符合我前几篇的“数据驱动框架”设计的问题;

消息队列

消息队列英语Message queue)是一种进程间通信或同一进程的不同线程间的通信方式,软件贮列用来处理一系列的输入,通常是来自使用者。消息队列提供了异步通信协议,每一个贮列中的纪录包含详细说明的资料,包含发生的时间,输入装置的种类,以及特定的输入参数,也就是说:消息的发送者和接收者不需要同时与消息队列互交。消息会保存在队列中,直到接收者取回它。

简单的说队列就是贮存了我们需要处理的Command但是并不是及时的拿到其处理结果;

实现

实际上,消息队列常常保存在链表结构中。拥有权限的进程可以向消息队列中写入或读取消息。

目前,有很多消息队列有很多开源的实现,包括JBoss MessagingJORAMApache ActiveMQSun Open Message QueueApache Qpid和HTTPSQS。

优点,缺点

消息队列本身是异步的,它允许接收者在消息发送很长时间后再取回消息,这和大多数通信协议是不同的。例如WWW中使用的HTTP协议是同步的,因为客户端在发出请求后必须等待服务器回应。然而,很多情况下我们需要异步的通信协议。比如,一个进程通知另一个进程发生了一个事件,但不需要等待回应。但消息队列的异步特点,也造成了一个缺点,就是接收者必须轮询消息队列,才能收到最近的消息。

信号相比,消息队列能够传递更多的信息。与管道相比,消息队列提供了有格式的数据,这可以减少开发人员的工作量。但消息队列仍然有大小限制。

读取队列消息

主要有两种(1)服务端的推;(2)客户端的拉;

拉:主要是客户端定时轮询拿走消息处理;

推:通过事件订阅方式主动通知订阅者进行处理;

消息的贮存

简单的是通过内存链表实现贮存;也可以借助DB,比如Redis;还可以持久到本地文件中;

如何保证异步处理的一致性

尽管队列主要目的是实现消息贮存,同时将调用与实现异步化。但是如果想达到处理消息一致性,好的方式是区别业务处理顺序,比如操作主从DB,主负责写,从负责读,我们没有机会在写之后立马从读数据库拿到你想要的结果;同时我们需要借助中间状态,当多个中间状态同时符合调用结果才到到业务时间被处理,否则将“异常消息”持久化,待下次操作;

上代码

建立消息对立核心队列

{
    public delegate void MessageQueueEventNotifyHandler(Message.BaseMessage message);

    public class MessageQueue:Queue<BaseMessage>
    {
        public static MessageQueue GlobalQueue = new MessageQueue();

        private Timer timer = new Timer();
        public MessageQueue() {
            this.timer.Interval = 5000;
            this.timer.Elapsed += Notify;
            this.timer.Enabled = true;
        }
        private void Notify(object sender, ElapsedEventArgs e) {
            lock (this) {
                if (this.Count > 0) {
                    //this.messageNotifyEvent.GetInvocationList()[0].DynamicInvoke(this.Dequeue());
                    var message = this.Dequeue();
                    this.messageNotifyEvent(message);
                }
            }
        }

        private MessageQueueEventNotifyHandler messageNotifyEvent;
        public event MessageQueueEventNotifyHandler MessageNotifyEvent {
            add {
                this.messageNotifyEvent += value;
            }

            remove {
                if (this.messageNotifyEvent != null) {
                    this.messageNotifyEvent -= value;
                }
            }
        }
    }
}

事件处理

public const string OrderCodePrefix = "P";
        public void Submit(Message.BaseMessage message)
        {
            Order order = message.Body as Order;

            if (order.OrderCode.StartsWith(OrderCodePrefix))
            {
                System.Console.WriteLine("这个是个正确的以({0})开头的订单:{1}", OrderCodePrefix,order.OrderCode);
            }
            else {
                System.Console.WriteLine("这个是个错误的订单,没有以({0})开头:{1}",OrderCodePrefix,order.OrderCode);
            }
        }

可依据具体业务进行个性化处理;

通过Proxy向队列追加消息

public class OrderServiceProxy:IOrderService
    {
        public void Submit(Message.BaseMessage message)
        {
            MessageQueue.MessageQueue.GlobalQueue.Enqueue(message);
        }
    }

客户端调用

OrderService orderService = new OrderService();
            MessageQueue.MessageQueue.GlobalQueue.MessageNotifyEvent += orderService.Submit;

            var orders = new List<Order>() { 
                new Order(){OrderCode="P001"},
                new Order(){OrderCode="P002"},
                new Order(){OrderCode="B003"}
            };

            OrderServiceProxy proxy = new OrderServiceProxy();
            orders.ForEach(order => proxy.Submit(new Message.BaseMessage() { Body=order}));

            Console.ReadLine();

这样就满足了事件的绑定与触发个性化处理,同时达到了消息异步化的目的,希望更细致的拓展用到后期的项目中。

目录
相关文章
|
6月前
|
编译器 数据处理 C#
C#中的异步流:使用IAsyncEnumerable<T>和await foreach实现异步数据迭代
【1月更文挑战第10天】本文介绍了C#中异步流的概念,并通过使用IAsyncEnumerable<T>接口和await foreach语句,详细阐述了如何异步地迭代数据流。异步流为处理大量数据或需要流式处理数据的场景提供了一种高效且非阻塞性的方法,使得开发者能够更优雅地处理并发和数据流问题。
|
2天前
|
C# UED SEO
C# 异步方法async / await任务超时处理
通过使用 `Task.WhenAny`和 `Task.Delay`方法,您可以在C#中有效地实现异步任务的超时处理机制。这种方法允许您在指定时间内等待任务完成,并在任务超时时采取适当的措施,如抛出异常或执行备用操作。希望本文提供的详细解释和代码示例能帮助您在实际项目中更好地处理异步任务超时问题,提升应用程序的可靠性和用户体验。
11 3
|
5月前
|
消息中间件 测试技术 RocketMQ
消息队列 MQ产品使用合集之在异步发送消息函数sendMessage()中出现了错误,错误代码为-3,该如何解决
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
4月前
|
消息中间件 存储 中间件
云原生异步问题之消息队列中的异步如何解决
云原生异步问题之消息队列中的异步如何解决
|
3月前
|
C#
C# async await 异步执行方法
C# async await 异步执行方法
51 0
|
5月前
|
C#
蓝易云 - C#将异步改成同步方法
注意:虽然这样可以将异步方法转为同步,但在实际开发中,我们通常推荐使用异步方法,因为它可以提高应用程序的响应性和并发性。将异步方法转为同步可能会导致死锁或性能问题。
43 2
|
5月前
|
消息中间件 存储 中间件
中间件消息队列协议异步通信
【6月更文挑战第5天】
44 2
|
6月前
|
消息中间件 存储 物联网
AMQP(Advanced Message Queuing Protocol)是一种标准化的消息队列协议,用于异步通信和消息传递
AMQP是开放标准的消息队列协议,确保异步通信中的消息路由、可靠性和灵活性。它支持持久化、确认机制,防止消息丢失,允许灵活路由与定制,适用于多领域,如企业应用、云计算和物联网。
140 0
|
6月前
|
C#
C#同步异步详解
C#同步异步详解
48 0
|
C# 开发者
C# 开发者技术:进程间数据共享之管道(Pipes)-异步通信版
主要类 1.NamedPipeClientStream 2.NamedPipeServerStream 解释:命名管道是一种进程间通信的方式,它允许不同进程之间在同一台机器上进行通信
919 2
C# 开发者技术:进程间数据共享之管道(Pipes)-异步通信版
下一篇
无影云桌面