《ASP.NET Core技术内幕与项目实战》精简集-DDD准备5.5：集成事件RabbitMQ

//声明RabbitMQ服务器，案例中以开发本机作为服务器
var factory = new ConnectionFactory();
factory.HostName = "localhost";
factory.DispatchConsumersAsync = true;

//命名交换机和路由名称，路由routinKey本质是一个事件,消费端通过routingKey来匹配队列
string exchangeName = "exchange1";
string routingKey = "event1";

//声明TCP连接，多个信道复用这个连接
using var conn = factory.CreateConnection();

//创建虚拟信道、连接交换机，发布信息
//本案例中，每延迟1秒，就发送一次信息
while (true)
{
    //定义传递的信息，每次发关事件
    string msg = DateTime.Now.TimeOfDay.ToString();

    //创建一个虚拟信道，连接交换机，发送消息
    using (var channel = conn.CreateModel())
    {
        //准备①：配置信道属性
        var properties = channel.CreateBasicProperties();
        properties.DeliveryMode = 2;

        //准备②：声明交接机。本例中，每次新的信道都使用相同的交接机
        channel.ExchangeDeclare(exchange:exchangeName,type:"direct");

        //准备③：RabbitMQ中的消息都是按照byte[]来传递，将消息转为byte[]格式
        byte[] body = Encoding.UTF8.GetBytes(msg);

        //发布消息，参数分别为交换机、routingKey（事件）、消息、信息属性
        channel.BasicPublish(exchange:exchangeName,routingKey:routingKey,body:body,basicProperties:properties,mandatory:true);
    }

    //每次发送完信息后，都关闭信道
    Console.WriteLine($"发布了消息：{msg}");

    //暂停1秒后，循环发送消息
    Thread.Sleep(1000);
}

//声明RabbitMQ服务器，案例中以开发本机作为服务器
var factory = new ConnectionFactory();
factory.HostName = "127.0.0.1";
factory.DispatchConsumersAsync = true;

//定义交换机和路由名称
string exchangeName = "exchange1";
string routingKey = "event1";

//声明TCP连接，多个信道复用这个连接
using var conn = factory.CreateConnection();

//声明一个信道
using var channel = conn.CreateModel();

//声明交换机
channel.ExchangeDeclare(exchange:exchangeName,type:"direct");

//声明队列
string queueName = "queue1";
channel.QueueDeclare(queue:queueName,durable:true,exclusive:false,autoDelete:false,arguments:null);

//将队列和指定交换机及routingKey绑定
channel.QueueBind(queue: queueName, exchange:exchangeName,routingKey:routingKey);

//通过信道，从队列中接收消息
var consumer = new AsyncEventingBasicConsumer(channel);

//定义一个事件处理函数
consumer.Received += ConsumerReceived ;

//执行事件处理
channel.BasicConsume(queue:queueName,autoAck:false,consumer:consumer)；
Console.ReadLine();

//收到消息后的事件处理函数
async Task ConsumerReceived(object sender,BasicDeliverEventArgs args)
{
    try
    {
        var bytes = args.Body.ToArray();
        string msg = Encoding.UTF8.GetString(bytes);
        Console.WriteLine(DateTime.Now + "收到了消息" + msg);
        channel.BasicAck(args.DeliveryTag, multiple: false); //通知队外，事件处理成功
        await Task.Delay(1000);
    }
    catch (Exception ex)
    {
        channel.BasicReject(args.DeliveryTag, true); //通知队列，事件处理失败，请重发
        Console.WriteLine("处理消息出错"+ex);
    }
}