.netcore3.1 RabbitMq Topics交换机

• .netcore3.1 RabbitMq Topics交换机
  • Topic Exchange介绍
  • Topic Exchange创建
  • Topic 队列的创建与消费
  • 运行结果

Topic Exchange介绍

Topic Exchange创建

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        channel.ExchangeDeclare(exchange: "MyTopicExchange", ExchangeType.Topic, true, false);
        var routingKey = (args.Length > 0) ? args[0] : "重庆市.南岸区.弹子石";
        var message = (args.Length > 1) ? string.Join(" ", args.Skip(1).ToArray()) : "Hello World!";
        var body = Encoding.UTF8.GetBytes(message);
        var properties = channel.CreateBasicProperties();
        properties.Persistent = true;
        channel.BasicPublish(exchange: "MyTopicExchange", routingKey, properties, body);
        Console.WriteLine(" [x] Sent '{0}':'{1}'", routingKey, message);
    }
}

Topic 队列的创建与消费

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
Console.WriteLine("Please enter a queue name:");
// MyTopicExchange1 MyTopicExchange2 
string queuename = Console.ReadLine();
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        channel.ExchangeDeclare(queuename, ExchangeType.Topic, true, false);
        // 声明一个队列
        channel.QueueDeclare(
            queue: queuename,
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        foreach (var bindingKey in args)
        {
            // 绑定队列相关正则匹配
            channel.QueueBind(queuename, "MyTopicExchange", bindingKey);
        }
        // 创建消费实例
        var consumer = new EventingBasicConsumer(channel);
        // 事件在交付到使用者时触发。（消费处理事件）
        consumer.Received += (model, ea) =>
        {
            string EventQueueName = queuename;
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            Console.WriteLine($" RountingKey: {ea.RoutingKey} Queue:{EventQueueName} Message: {message}");
            // 手动确认
            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: true);
        };
        // 绑定到队列中去
        channel.BasicConsume(queue: queuename, autoAck: false, consumer: consumer);
        Console.ReadLine();
    }
}

运行结果