RabbitMQ与.net core(二)Producer与Exchange

Producer：消息的生产者，也就是创建消息的对象

Exchange：消息的接受者，也就是用来接收消息的对象，Exchange接收到消息后将消息按照规则发送到与他绑定的Queue中。下面我们来定义一个Producer与Exchange。

1.新建.netcore console项目，并引入RabbitMQ.Client的Nuget包

2.创建Exchange

using RabbitMQ.Client;

namespace RabbitMQConsole

{

    class Program

    {

        static void Main(string[] args)

        {

            ConnectionFactory factory = new ConnectionFactory();

            factory.HostName = "39.**.**.**";

            factory.Port = ;

            factory.VirtualHost = "/";

            factory.UserName = "root";

            factory.Password = "root";

            var exchange = "change2";

            var route = "route2";

            var queue = "queue2";

            using (var connection = factory.CreateConnection())

            {

                using (var channel = connection.CreateModel())

                {

                    channel.ExchangeDeclare(exchange, type:"direct", durable: true, autoDelete: false);   //创建Exchange

                }

            }

        }

    }

}

可以看到Echange的参数有：

type：可选项为，fanout，direct，topic，headers。区别如下：

　　　　fanout：发送到所有与当前Exchange绑定的Queue中

　　　　direct：发送到与消息的routeKey相同的Rueue中

　　　　topic：fanout的模糊版本

　　　　headers：发送到与消息的header属性相同的Queue中

durable：持久化

autoDelete：当最后一个绑定（队列或者exchange）被unbind之后，该exchange自动被删除。

运行程序，可以在可视化界面看到change2

接下来我们可以创建与change2绑定的queue

3.创建Queue

                using (var channel = connection.CreateModel())

                {

                    channel.ExchangeDeclare(exchange, type: "direct", durable: true, autoDelete: false);

                    channel.QueueDeclare(queue, durable: true, exclusive: false, autoDelete: false);  #创建queue2

                    channel.QueueBind(queue, exchange, route);　　#将queue2绑定到exchange2

                }

可以看到Echange的参数有：

durable：持久化

exclusive：如果为true，则queue只在channel存在时存在，channel关闭则queue消失

autoDelete：当最后一个绑定（队列或者exchange）被unbind之后，该exchange自动被删除。

去可视化界面看Queue

4.发送消息

                using (var channel = connection.CreateModel())

                {

                    channel.ExchangeDeclare(exchange, type: "direct", durable: true, autoDelete: false);

                    channel.QueueDeclare(queue, durable: true, exclusive: false, autoDelete: false);

                    channel.QueueBind(queue, exchange, route);

                    var props = channel.CreateBasicProperties();

                    props.Persistent = true; #持久化

                    channel.BasicPublish(exchange, route, true, props, Encoding.UTF8.GetBytes("hello rabbit"));

                }

5.消费消息

using RabbitMQ.Client;

using System;

using System.Text;

namespace RabbitMQClient

{

    class Program

    {

        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory()

        {

            HostName = "39.**.**.**",

            Port = ,

            UserName = "root",

            Password = "root",

            VirtualHost = "/"

        };

        static void Main(string[] args)

        {

            var exchange = "change2";

            var route = "route2";

            var queue = "queue2";

            using (IConnection conn = rabbitMqFactory.CreateConnection())

            using (IModel channel = conn.CreateModel())

            {

                channel.ExchangeDeclare(exchange, "direct", durable: true, autoDelete: false);

                channel.QueueDeclare(queue, durable: true, exclusive: false, autoDelete: false);

                channel.QueueBind(queue, exchange, route);

                while (true)

                {

                    var message = channel.BasicGet(queue, true);  #第二个参数说明自动释放消息，如为false需手动释放消息

                    if(message!=null)

                    {

                        var msgBody = Encoding.UTF8.GetString(message.Body);

                        Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));

                    }

                    System.Threading.Thread.Sleep(TimeSpan.FromSeconds());

                }

            }

        }

    }

}

运行查看结果

查看可视化界面

6.手动释放消息

                while (true)

                {

                    var message = channel.BasicGet(queue, false);#设置为手动释放

                    if(message!=null)

                    {

                        var msgBody = Encoding.UTF8.GetString(message.Body);

                        Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));

                    }

                    channel.BasicAck(message.DeliveryTag, false); #手动释放

                    System.Threading.Thread.Sleep(TimeSpan.FromSeconds());

                }

我们再发一条消息，然后开始消费，加个断点调试一下

查看一下Queue中消息状态

然后直接取消调试，不让程序走到释放的那一步，再查看一下消息状态

这么说来只要不走到 channel.BasicAck(message.DeliveryTag, false);这一行，消息就不会被释放掉，我们让程序直接走到这一行代码，查看一下消息的状态

如图已经被释放了

7.让失败的消息回到队列中

                while (true)

                {

                    var message = channel.BasicGet(queue, false);

                    if(message!=null)

                    {

                        var msgBody = Encoding.UTF8.GetString(message.Body);

                        Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));

                        Console.WriteLine(message.DeliveryTag);　　　　#当前消息被处理的次序数

                        if (==)

                            channel.BasicReject(message.DeliveryTag, true);

                    }

                    System.Threading.Thread.Sleep(TimeSpan.FromSeconds());

                }

重新发送4条消息

开始消费

我们可以看到消息一直没有没消费，因为消息被处理之后又放到了队尾

8.监听消息

 using (IConnection conn = rabbitMqFactory.CreateConnection())

            using (IModel channel = conn.CreateModel())

            {

                channel.ExchangeDeclare(exchange, "direct", durable: true, autoDelete: false);

                channel.QueueDeclare(queue, durable: true, exclusive: false, autoDelete: false);

                channel.QueueBind(queue, exchange, route);

                channel.BasicQos(prefetchSize: , prefetchCount: 1, global: false);  #一次接受10条消息，否则rabbit会把所有的消息一次性推到client，会增大client的负荷

                EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

                consumer.Received += (model, ea) =>

                {

                    Byte[] body = ea.Body;

                    String message = Encoding.UTF8.GetString(body);

                    Console.WriteLine( message+Thread.CurrentThread.ManagedThreadId);

                    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);

                };

                channel.BasicConsume(queue: queue, autoAck: false, consumer: consumer);

                Console.ReadLine();

            }