基于NET Framework使用阿里云AMQP

概述

消息队列 AMQP 由阿里云消息队列（MQ）团队基于 AMQP 0.91 标准协议研发，完全兼容 RabbitMQ 开源社区，打造分布式、高吞吐、低延迟、高可扩展的云消息服务。用户可开箱即用，无需部署免运维，从而实现快速上云，阿里云提供全托管服务，更专业、更可靠、更安全。本文主要演示如何在NET Framework环境下使用阿里云AMQP服务。

使用

1、NuGet 安装 RabbitMQ.Client

2、使用的NET Framework版本为4.6.0

3、认证需要的工具类Utils参考链接。

4、发送端示例代码

using System;
using System.Collections.Generic;
using System.Text;
using RabbitMQ.Client;
using System.Threading;
using RabbitMQ.Client.Exceptions;

namespace amqp_dotnetframework_demo
{
    class Program
    {
        static void Main(string[] args) { 

            //测试队列的名称
            string QueueName = "helloworldqueue";

            var factory = new ConnectionFactory();
            /*接入点*/
            factory.HostName = "********";
            /*阿里云的accessKey*/
            factory.UserName = "********";
            /*阿里云的accessSecret*/
            factory.Password = "********";
            //虚拟主机的名称，需要提前在管理门户创建
            factory.VirtualHost = "********"; 
            /*默认端口*/
            factory.Port = 5672;
            factory.AuthMechanisms = new List<AuthMechanismFactory>() { new AliyunMechanismFactory() };
            factory.TopologyRecoveryEnabled = true;

            ConnectionFactory rabbitMqFactory = factory;

            // 发送消息测试
            var connection = factory.CreateConnection();
            var channel = connection.CreateModel();
            channel.QueueDeclare(queue: QueueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
            while (true)
            {
                try
                {
                    if (!connection.IsOpen)
                    {
                        connection.Close();
                        connection = factory.CreateConnection();
                        channel = connection.CreateModel();
                    }
                    string message = Guid.NewGuid().ToString();
                    var body = Encoding.UTF8.GetBytes(message);

                    channel.BasicPublish(exchange: "", routingKey: QueueName, basicProperties: null, body: body);
                    Console.WriteLine(" [x] Sent {0}", message);
                    Thread.Sleep(1000);
                }
                catch (BrokerUnreachableException e)
                {
                    Thread.Sleep(3000);
                    Console.WriteLine(e);
                    continue;
                }
                catch (Exception e)
                {
                    Console.WriteLine(e);
                    continue;
                }
            }
            Console.ReadKey();
        }
    }
}

5、接收端示例代码

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

namespace amqp_dotnetframework_demo
{
    class Program
    {
        static void Main(string[] args) { 

            //测试队列的名称
            string QueueName = "helloworldqueue";

            var factory = new ConnectionFactory();
            /*接入点*/
            factory.HostName = "********";
            /*阿里云的accessKey*/
            factory.UserName = "********";
            /*阿里云的accessSecret*/
            factory.Password = "********";
            //虚拟主机的名称，需要提前在管理门户创建
            factory.VirtualHost = "********"; 
            /*默认端口*/
            factory.Port = 5672;
            factory.AuthMechanisms = new List<AuthMechanismFactory>() { new AliyunMechanismFactory() };
            factory.TopologyRecoveryEnabled = true;

            ConnectionFactory rabbitMqFactory = factory;

            // 消费消息测试
            using (var connection = rabbitMqFactory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                //channel.QueueDeclare(queue: "queueNewDemo", durable: true, exclusive: false, autoDelete: false, arguments: null);

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

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

            Console.ReadKey();
        }
    }
}

6、接收端测试情况