C# 基于消息发布订阅模型的示例(下)

我来为您详细讲解一下“C# 基于消息发布订阅模型的示例(下)” 的完整攻略。

什么是消息发布订阅模型？

消息发布订阅模型是一种消息传递模式，它包含一个发布者和许多订阅者。发布者通常不知道有哪些订阅者，而订阅者则可以动态地订阅和取消订阅发布者。

消息发布订阅模型的优势

• 较容易增加或删除订阅者，无需修改发布者代码。
• 可以实现一对多的关系，即一个事件可以有多个订阅者。
• 发布者和订阅者之间松散耦合，使得系统更易于扩展和维护。

示例说明一：使用 RabbitMQ 实现基于消息发布订阅模型

要使用 RabbitMQ 实现消息发布订阅模型，需要引入 RabbitMQ 的消息中间件，并编写发布者和订阅者的代码。

首先需要创建一个名为 RabbitMQExample 的控制台应用程序，引入 RabbitMQ 的 NuGet 包。代码如下：

using RabbitMQ.Client;
using System;
using System.Text;

namespace RabbitMQExample
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: "logs", type: "fanout");

                var message = GetMessage(args);
                var body = Encoding.UTF8.GetBytes(message);
                channel.BasicPublish(exchange: "logs",
                                     routingKey: "",
                                     basicProperties: null,
                                     body: body);
                Console.WriteLine(" [x] Sent {0}", message);
            }

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }

        private static string GetMessage(string[] args)
        {
            return ((args.Length > 0)
                   ? string.Join(" ", args)
                   : "info: Hello World!");
        }
    }
}

这是发布者的代码，它将消息发布到指定的交换机。在这里，我们创建了一个名为 logs 的交换机，类型为 fanout。其他类型的交换机有 direct、topic 和 headers，但相较之下，fanout 类型的交换机不处理路由键而是简单地将消息广播到所有队列中。

然后，我们需要编写订阅者的代码。代码如下：

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;

namespace RabbitMQExample
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange: "logs", type: "fanout");

                var queueName = channel.QueueDeclare().QueueName;
                channel.QueueBind(queue: queueName,
                                  exchange: "logs",
                                  routingKey: "");

                Console.WriteLine(" [*] Waiting for logs.");

                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body.ToArray();
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine(" [x] {0}", message);
                };
                channel.BasicConsume(queue: queueName,
                                     autoAck: true,
                                     consumer: consumer);

                Console.WriteLine(" Press [enter] to exit.");
                Console.ReadLine();
            }
        }
    }
}

在这里我们同样创建一个名为 logs 的交换机，并创建一个随机的队列，将队列绑定到交换机上。这个代码块向我们展示了一个用于订阅并处理这些日志消息的简单示例。

运行代码的方式是，首先开启一个控制台窗口作为发布者，并输入需要发布的消息，然后点击回车确认发送。然后再开启一个控制台窗口作为订阅者，订阅这些消息并执行相应的处理逻辑。

示例说明二：使用 SignalR 实现基于消息发布订阅模型

如果你使用的是 ASP.NET Core，SignalR 可能是最简单的解决方案，因为它不需要引入额外的消息中间件，而是使用了自带的信道和远程调用机制。在 ASP.NET Core 中使用 SignalR，你需要添加以下 NuGet 包：Microsoft.AspNetCore.SignalR 和 Microsoft.AspNetCore.SignalR.Core。

using Microsoft.AspNetCore.SignalR;
using System.Threading.Tasks;

namespace SignalRExample.Hubs
{
    public class ChatHub : Hub
    {
        public async Task SendMessage(string user, string message)
        {
            await Clients.All.SendAsync("ReceiveMessage", user, message);
        }
    }
}

在这里，我们创建了一个名为 ChatHub 的 SignalR hub。当一个用户发送消息时，我们将广播这条消息到所有已连接的客户端。

在这里，hub 继承了 SignalR 的 Hub 类，它包含了一些方便的方法来管理客户端和服务器之间的连接状态。这个类中有一个 SendMessage 方法，该方法接收一个用户名和消息作为参数，并对所有已连接的客户端发送消息。

下面是一个简单的客户端文件：

"use strict";

var connection = new signalR.HubConnectionBuilder().withUrl("/chatHub").build();

connection.on("ReceiveMessage", function (user, message) {
    var encodedMsg = user + " says " + message;
    var li = document.createElement("li");
    li.textContent = encodedMsg;
    document.getElementById("messagesList").appendChild(li);
});

connection.start().then(function () {
    document.getElementById("sendButton").addEventListener("click", function (event) {
        var user = document.getElementById("userInput").value;
        var message = document.getElementById("messageInput").value;
        connection.invoke("SendMessage", user, message).catch(function (err) {
            return console.error(err.toString());
        });
        event.preventDefault();
    });
}).catch(function (err) {
    return console.error(err.toString());
});

这个 JavaScript 文件定义了一个名为 connection 的 HubConnection 对象，该对象用于与 ChatHub 进行通信。当一个用户发送消息时，所有已连接的客户端都会接收到该消息，并在网页上显示。

要使用此示例，你需要在 Startup.cs 文件中添加以下代码：

app.UseEndpoints(endpoints =>
{
    endpoints.MapHub<ChatHub>("/chatHub");
});

在这个文件里，你需要添加 SignalR 中间件以配置 SignalR 端点。通过以下方式注册端点：

endpoints.MapHub<ChatHub>("/chatHub");

现在你可以运行你的 ASP.NET Core 应用程序，并打开两个浏览器窗口以模拟两个用户。在其中一个窗口中发送一条消息，你将会在两个窗口都看到这条消息的出现。

这些示例为你提供了一些基本的实现。不同的开发场景需要使用不同的解决方案。在实现消息发布订阅模型时，有许多其他可供选择的方案，例如：Windows Workflow Foundation、Apache Kafka、Azure Service Bus 等。您可以根据自己的需求和实际情况选择其中最适合的方案。