RabbitMQ七种工作模式

标签: rabbitmq

概念

   ☞ RabbitMQ的一个消息代理:它接受并转发消息。
   ☞ 你可以把它当成一个邮局:当你想邮寄信件的时候,你会把信件放在投 
   递箱中,并确信邮递员最终会将信件送到收件人的手里。在这个例子中,
   RabbitMQ就相当于投递箱、邮局和邮递员。
   ☞ AMQP协议中的核心思想就是生产者和消费者隔离,生产者从不直接将消
   息发送给队列。生产者通常不知道是否一个消息会被发送到对队列中,只是
   将消息发送到一个交换机(Exchange)。先由交换机(Exchange)来接
   收,然后交换机(Exchange)按照特定的策略转发Queue进行存储。同
   理,消费者也是如此。交换机(Exchange),转发各个消息分发到对应的
   队列。
  (参考链接:https://blog.csdn.net/qq_33040219/article/details/82383127)

一. simple:简单模式

1. 要点

一个生产者,一个消费者。

2. 原理

simple模式

☞ 消息发布者(Publish)将消息放入队列(默认交换机)。
☞ 消息的消费者(Consumer)监听(while)消息队列,如果队列中有消
息,就消费掉,消息被拿走后,自动从队列中删除(隐患:消息可能没有被
消费者正确处理,已经从队列中消失了,造成消息的丢失)。应用场景:聊
天(中间有一个过度的服务器;P端,C端)。

3. 代码演示

3.1 发布端(Publish)

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace KibaRabbitMQSend
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
            factory.UserName = "guest";//默认用户名,用户可以在服务端自定义创建,有相关命令行
            factory.Password = "guest";//默认密码
            using (var connection = factory.CreateConnection())//连接服务器,即正在创建终结点。
            {
                //创建一个通道,这个就是Rabbit自己定义的规则了,如果自己写消息队列,这个就可以开脑洞设计了
                //这里Rabbit的玩法就是一个通道channel下包含多个队列Queue
                using (var channel = connection.CreateModel())
                {
                    //                               持久化 
                    channel.QueueDeclare("kibaQueue", false, false, false, null);//创建一个名称为kibaqueue的消息队列
                    var properties = channel.CreateBasicProperties();
                    //properties.DeliveryMode = 1;//1:非持久化;2:持久化
                    string message = "I am Kiba518"; //传递的消息内容
                    channel.BasicPublish(string.Empty, "kibaQueue", null, Encoding.UTF8.GetBytes(message)); //生产消息
                    Console.WriteLine($"Send:{message}");
                    Console.ReadKey();
                }
            }
        }
    }
}

启动成功后,消息被发布,如图:
消息发布成功

3.2 消费端(Consumer)

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace KibaRabbitMQReceived
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";
            factory.UserName = "guest";
            factory.Password = "guest";

            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.QueueDeclare("kibaQueue", false, false, false, null);

                    /* 这里定义了一个消费者,用于消费服务器接受的消息
                     * C#开发需要注意下这里,在一些非面向对象和面向对象比较差的语言中,是非常重视这种设计模式的。
                     * 比如RabbitMQ使用了生产者与消费者模式,然后很多相关的使用文章都在拿这个生产者和消费者来表述。
                     * 但是,在C#里,生产者与消费者对我们而言,根本算不上一种设计模式,他就是一种最基础的代码编写规则。
                     * 所以,大家不要复杂的名词吓到,其实,并没那么复杂。
                     * 这里,其实就是定义一个EventingBasicConsumer类型的对象,然后该对象有个Received事件,
                     * 该事件会在服务接收到数据时触发。
                     */
                    var consumer = new EventingBasicConsumer(channel);//消费者
                    channel.BasicConsume("kibaQueue", true, consumer);//消费消息
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body;
                        var message = Encoding.UTF8.GetString(body);
                    };
                }
            }
            Console.WriteLine($"消息接收完成!");
            Console.ReadKey();
        }
    }
}

启动成功后,消息被消费,如图:
消息被消费

二. work:工作模式(资源的竞争)

1. 要点

一个生产者,多个消费者,消息被多个消费者竞争接收。

2. 原理

工作模式
☞ 消息产生者将消息放入队列消费者可以有多个,消费者1,消费者2,同时监听
同一个队列,消息被消费。C1 C2共同争抢当前的消息队列内容,谁先拿到谁负责
消费消息(隐患:高并发情况下,默认会产生某一个消息被多个消费者共同使用,
可以设置一个开关(syncronize,与同步锁的性能不一样) 保证一条消息只能被
一个消费者使用)。
☞ 应用场景:红包;大项目中的资源调度(任务分配系统不需知道哪一个任务执
行系统在空闲,直接将任务扔到消息队列中,空闲的系统自动争抢)。

3. 代码演示

3.1 发布端(Publish)

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Publish
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };

            #region 1. 发布/订阅方式,多个客户端(消费者)竞争处理消息,减轻单个客户端的负载压力
            //第一步:创建connection
            using (var connection = factory.CreateConnection())//连接服务器,即正在创建终结点。
            {
                //第二步:创建一个channel
                using (var channel = connection.CreateModel())
                {
                    //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange,可以不用自己定义,定义了,一定要绑定 队列和交换机)
                    channel.ExchangeDeclare("myexchange", ExchangeType.Direct, true, false,null);

                    //第四步:创建一个队列(queue)
                    channel.QueueDeclare("mytest", true, false, false, null);

                    channel.QueueBind("mytest", "myexchange", "mytest", null);//交换机绑定队列

                    for (int i = 0; i < 100; i++)
                    {
                        var msg = Encoding.UTF8.GetBytes(string.Format("{0}:{1}", i, "您好!"));
                        //第五步:发布消息
                        channel.BasicPublish(string.Empty, routingKey: "mytest", basicProperties: null, body: msg);
                    }
                    Console.WriteLine($"Send End");
                    Console.ReadKey();
                }
            }
            #endregion 
        }
    }
}

3.2 消费端(Consumer)

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.MessagePatterns;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Consumer01
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };

            #region 1. 发布/订阅方式,多个客户端(消费者)竞争处理消息,减轻单个客户端的负载压力

            //第一步:创建connection
            var connection = factory.CreateConnection();

            //第二步:创建一个channel
            var channel = connection.CreateModel();

            //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange)

            //第四步:声明队列
            channel.QueueDeclare("mytest", true, false, false, null);

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                var msg = Encoding.UTF8.GetString(e.Body);
                Console.WriteLine(msg);
            };
            channel.BasicConsume("mytest", true, consumer);
            #endregion
            Console.ReadKey();
        }
    }
}

先启动两个消费端,在启动发布端。发现两客户端平均分摊了消息,如图:
work模式

三. publish/subscribe:发布订阅(共享资源)

1. 要点

一个生产者,多个消费者,消息复制多份,每个消费者接收相同的消息。

2. 原理

发布订阅模式
☞ X代表交换机RabbitMQ内部组件,erlang 消息产生者是代码完
成,代码的执行效率不高,消息产生者将消息放入交换机,交换机发
布订阅把消息发送到所有消息队列中,对应消息队列的消费者拿到 消息进行消费
☞相关场景:邮件群发,群聊天,广播(广告)

3. 代码演示

3.1 发布端(Publish)

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Publish
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region  3. 发布/订阅方式,多个客户端(消费者)复制消费信息,得到数据是一摸一样的。
            //第一步:创建connection
            using (var connection = factory.CreateConnection())//连接服务器,即正在创建终结点。
            {
                //第二步:创建一个channel
                using (var channel = connection.CreateModel())
                {
                    for (int i = 0; i < 100; i++)
                    {
                        var msg = Encoding.UTF8.GetBytes(string.Format("{0}:{1}", i, "您好!"));
                        //第五步:发布消息
                        channel.BasicPublish("myfanoutexchange", routingKey: string.Empty, basicProperties: null, body: msg);
                        Console.WriteLine(i);
                    }
                    Console.ReadKey();
                }
            }
            #endregion
        }
    }
}

3.2 消费端(Consumer)

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.MessagePatterns;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Consumer01
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region 3. 发布/订阅方式,多个客户端(消费者)复制消费信息,得到数据是一摸一样的。。
            //第一步:创建connection
            var connection = factory.CreateConnection();

            //第二步:创建一个channel
            var channel = connection.CreateModel();

            //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange)
            channel.ExchangeDeclare("myfanoutexchange", ExchangeType.Fanout, true, false, null);

            //第四步:声明队列
            channel.QueueDeclare("myfanoutqueue", true, false, false, null);

            //将myfanoutqueue绑定到交换机"myfanoutexchange"中
            channel.QueueBind("myfanoutqueue", "myfanoutexchange", string.Empty, null);

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                var msg = Encoding.UTF8.GetString(e.Body);
                Console.WriteLine(msg);
            };
            channel.BasicConsume("myfanoutqueue", true, consumer);
            Console.WriteLine("consumer1 端启动完毕");
            #endregion
            Console.ReadKey();
        }
    }
}
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Consumer02
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region 3. 发布/订阅方式,多个客户端(消费者)复制消费信息,得到数据是一摸一样的。。
            //第一步:创建connection
            var connection = factory.CreateConnection();

            //第二步:创建一个channel
            var channel = connection.CreateModel();

            //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange)
            channel.ExchangeDeclare("myfanoutexchange", ExchangeType.Fanout, true, false, null);

            //第四步:声明队列
            channel.QueueDeclare("myfanoutqueue2", true, false, false, null);

            //将myfanoutqueue2绑定到交换机"myfanoutexchange"中
            channel.QueueBind("myfanoutqueue2", "myfanoutexchange", string.Empty, null);

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                var msg = Encoding.UTF8.GetString(e.Body);
                Console.WriteLine(msg);
            };
            channel.BasicConsume("myfanoutqueue2", true, consumer);
            Console.WriteLine("consumer2 端启动完毕");
            #endregion
            Console.ReadKey();
        }
    }
}

先后启动两消费端,再骑启动发布端。发现两消费端接收的消息是一样的。如图:
发布订阅

四. routing:路由模式

1. 要点

一个生产者,多个消费者,多个消费者根据路由类型分摊处理消息,减轻单
个客户端的负载压力【场景:日志(info、debug、error、warring...)】。

2. 原理

路由模式
☞ 消息生产者将消息发送给交换机按照路由判断,路由是字符串(info) 当前产生的消息携带路由字符(对象的方法),交换机根据路由的key,只能匹配上路由key对应的消息队列,对应的消费者才能消费消息。
☞ 根据业务功能定义路由字符串。
☞ 从系统的代码逻辑中获取对应的功能字符串,将消息任务扔到对应的队列中业务场景:error 通知;EXCEPTION;错误通知的功能;传统意义的错误通知;客户通知;利用key路由,可以将程序中的错误封装成消息传入到消息队列中,开发者可以自定义消费者,实时接收错误。

3. 代码演示

3.1 发布端(Publish)

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Publish
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region 2. 发布/订阅方式,多个客户端(消费者)根据信息类型分摊处理消息,减轻单个客户端的负载压力【场景:日志(info、debug、error、warring...)】
            //第一步:创建connection
            using (var connection = factory.CreateConnection())//连接服务器,即正在创建终结点。
            {
                //第二步:创建一个channel
                using (var channel = connection.CreateModel())
                {
                    for (int i = 0; i < 100; i++)
                    {
                        var msg = Encoding.UTF8.GetBytes(string.Format("{0}:{1}", i, "您好!"));
                        var level = i % 13 == 0 ? "info" : "error";
                        //第五步:发布消息
                        channel.BasicPublish("myexchange", routingKey: level, basicProperties: null, body: msg);
                        Console.WriteLine(i);
                    }
                    Console.ReadKey();
                }
            }
            #endregion
        }
    }
}

3.2 消费端(Consumer)

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.MessagePatterns;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Consumer01
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region 2. 发布/订阅方式,多个客户端(消费者)根据信息类型分摊处理消息,减轻单个客户端的负载压力【场景:日志(info、debug、error、warring...)】

            //第一步:创建connection
            var connection = factory.CreateConnection();

            //第二步:创建一个channel
            var channel = connection.CreateModel();

            //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange)
            channel.ExchangeDeclare("myexchange", ExchangeType.Direct, true, false, null);

            //第四步:声明队列
            channel.QueueDeclare("log_else", true, false, false, null);

            var arr = new string[3] { "debug", "info", "warning" };

            //将debug、info、warning都绑定到"log_else"队列中
            for (int i = 0; i < arr.Length; i++)
            {
                channel.QueueBind("log_else", "myexchange", arr[i], null);
            }

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                var msg = Encoding.UTF8.GetString(e.Body);
                Console.WriteLine(msg);
            };
            channel.BasicConsume("log_else", true, consumer);
            #endregion
            Console.ReadKey();
        }
    }
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Consumer02
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };

            #region 2. 发布/订阅方式,多个客户端(消费者)根据信息类型分摊处理消息,减轻单个客户端的负载压力【场景:日志(info、debug、error、warring...)】
            //第一步:创建connection
            var connection = factory.CreateConnection();

            //第二步:创建一个channel
            var channel = connection.CreateModel();

            //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange)
            channel.ExchangeDeclare("myexchange", ExchangeType.Direct, true, false, null);

            //第四步:声明队列
            channel.QueueDeclare("log_error", true, false, false, null);

            //将debug、info、warning都绑定到"log_else"队列中
            channel.QueueBind("log_error", "myexchange", "error", null);

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                var msg = Encoding.UTF8.GetString(e.Body);
                Console.WriteLine(msg);
            };
            channel.BasicConsume("log_error", true, consumer);
            #endregion
            Console.ReadKey();
        }
    }
}

先启动两个消费端,再启动发布端。发现消费端根据不同的路由,接收响应的消息。如图:
路由模式

五. topic:主题模式

1. 要点

一个生产者,多个消费者,多个消费者根据路由类型分摊处理消息,此处的路由的用*号做模糊匹配的。

2. 原理

主题模式
☞ 星号井号代表通配符。
☞ 星号代表多个单词,井号代表一个单词。
☞ 路由功能添加模糊匹配。
☞ 消息产生者产生消息,把消息交给交换机。
☞ 交换机根据key的规则模糊匹配到对应的队列,由队列的监听消费者接收消息消费。

3. 代码演示

3.1 发布端(Publish)

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Publish
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region  5. topic方式:模糊归类,被对应的消费者接收使用。
            //第一步:创建connection
            using (var connection = factory.CreateConnection())//连接服务器,即正在创建终结点。
            {
                //第二步:创建一个channel
                using (var channel = connection.CreateModel())
                {
                    for (int i = 0; i < 100; i++)
                    {
                        var msg = Encoding.UTF8.GetBytes(string.Format("{0}:{1}", i, "您好!"));
                        //第五步:发布消息
                        var routingKey = i % 13 == 0 ? "*.com" : "*.cn";
                        channel.BasicPublish("mytopicexchange", routingKey: routingKey, basicProperties: null, body: msg);
                        Console.WriteLine(i);
                    }
                    Console.ReadKey();
                }
            }
            #endregion
        }
    }
}

3.2 消费端(Consumer)

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.MessagePatterns;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Consumer01
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region 5. topic方式:模糊归类,被对应的消费者接收使用。
            //第一步:创建connection
            var connection = factory.CreateConnection();

            //第二步:创建一个channel
            var channel = connection.CreateModel();

            //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange)
            channel.ExchangeDeclare("mytopicexchange", ExchangeType.Topic, true, false, null);

            //第四步:声明队列

            channel.QueueDeclare("mytopicqueue", true, false, false, null);

            //将myfanoutqueue绑定到交换机"myfanoutexchange"中
            channel.QueueBind("mytopicqueue", "mytopicexchange", "*.com", null);

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                var msg = Encoding.UTF8.GetString(e.Body);
                Console.WriteLine(msg);
            };
            channel.BasicConsume("mytopicqueue", true, consumer);
            Console.WriteLine("consumer1 端启动完毕");
            #endregion
            Console.ReadKey();
        }
    }
}
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Consumer02
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region 5. topic方式:模糊归类,被对应的消费者接收使用。
            //第一步:创建connection
            var connection = factory.CreateConnection();

            //第二步:创建一个channel
            var channel = connection.CreateModel();

            //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange)
            channel.ExchangeDeclare("mytopicexchange", ExchangeType.Topic, true, false, null);

            //第四步:声明队列

            channel.QueueDeclare("mytopicqueue2", true, false, false, null);

            //将myfanoutqueue绑定到交换机"myfanoutexchange"中
            channel.QueueBind("mytopicqueue2", "mytopicexchange", "*.cn", null);

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                var msg = Encoding.UTF8.GetString(e.Body);
                Console.WriteLine(msg);
            };
            channel.BasicConsume("mytopicqueue2", true, consumer);
            Console.WriteLine("consumer2 端启动完毕");
            #endregion
            Console.ReadKey();
        }
    }
}

先启动两个消费端,再启动发布端。发现消费端根据不同的路由,接收响应的消息。如图:
主题模式

六. Header:头交换机模式

1. 要点

一个生产者,多个消费者,在消息头添加条件,匹配才被消费者接收使用(header Exchange类型用的比较少,但还是知道一点好)。

2. 原理

☞ header exchange(头交换机)和主题交换机有点相似,但是不同于主题交换
机的路由是基于路由键,头交换机的路由值基于消息的header数据。 
☞ 主题交换机路由键只有是字符串,而头交换机可以是整型和哈希值 。

3. 代码演示

3.1 发布端(Publish)

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Publish
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region  4. headers方式:在消息头添加条件,匹配才被消费者接收使用。
            //第一步:创建connection
            using (var connection = factory.CreateConnection())//连接服务器,即正在创建终结点。
            {
                //第二步:创建一个channel
                using (var channel = connection.CreateModel())
                {
                    var properties = channel.CreateBasicProperties();
                    properties.Headers = new Dictionary<string, object>();
                    properties.Headers.Add("username", "jack");
                    properties.Headers.Add("password", "12345");
                    for (int i = 0; i < 100; i++)
                    {
                        var msg = Encoding.UTF8.GetBytes(string.Format("{0}:{1}", i, "您好!"));
                        //第五步:发布消息
                        channel.BasicPublish("myheadersexchange", routingKey: string.Empty, basicProperties: properties, body: msg);
                        Console.WriteLine(i);
                    }
                    Console.ReadKey();
                }
            }
            #endregion
        }
    }
}

3.2 消费端(Consumer)

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.MessagePatterns;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Consumer01
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory()
            {
                HostName = "localhost",//主机名,Rabbit会拿这个IP生成一个endpoint,这个很熟悉吧,就是socket绑定的那个终结点。
                UserName = "guest",//默认用户名,用户可以在服务端自定义创建,有相关命令行
                Password = "guest"//默认密码
            };
            #region 4. headers方式:在消息头添加条件,匹配才被消费者接收使用。
            //第一步:创建connection
            var connection = factory.CreateConnection();

            //第二步:创建一个channel
            var channel = connection.CreateModel();

            //第三步:声明交换机(因为rabbitmq已经有了自定义的abpq default exchange)
            channel.ExchangeDeclare("myheadersexchange", ExchangeType.Headers, true, false, null);

            //第四步:声明队列
            Dictionary<string, object> dic = new Dictionary<string, object>();
            //dic.Add("x-match", "any");//满足任一条件即被消费者接收
            dic.Add("x-match", "all");//满足任所有条件才能被消费者接收
            dic.Add("username", "jack");//条件1
            dic.Add("password", "12345");//条件2
            channel.QueueDeclare("myheadersqueue", true, false, false, null);

            //将myfanoutqueue绑定到交换机"myfanoutexchange"中
            channel.QueueBind("myheadersqueue", "myheadersexchange", string.Empty, dic);

            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, e) =>
            {
                var msg = Encoding.UTF8.GetString(e.Body);
                Console.WriteLine(msg);
            };
            channel.BasicConsume("myheadersqueue", true, consumer);
            Console.WriteLine("consumer1 端启动完毕");
            #endregion
            Console.ReadKey();
        }
    }
}

先启动发布端,再启动消费端。发现消费端接收所有匹配消息头的消息。如图:
Header模式

七. RPC(暂留)

1. 要点

2. 原理

3. 代码演示

3.1 发布端(Publish)

版权声明:本文为weixin_40877204原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/weixin_40877204/article/details/103828076

智能推荐

java反编译

jvm 把Boolean类型的值flag当做int类型处理。​​​ Foo.java: 由 class 文件生成 jasm 文件:java -jar asmtools.jar jdis Foo.class > Foo.jasm  修改jasm文件: 执行反编译: java -jar jd-gui-1.6.6.jar File 打开Foo.class文件:b修改为2 重新执行java...

【学习笔记】03-v-html的学习和示例

v-html的认识和使用 示例: 显示结果: 注意:v-html是有复制的...

Java实现在线考试系统(系统介绍)

1.和现在有的考试系统有以下几种优势: a.和现在有的系统比较起来,本系统有科目、章节、老师、学生、班级等信息的管理,还有批阅试卷查看已批阅试卷等。传统的考试系统划分并不细,业务功能简单。 b.和学校的考试系统还有外面的考试系统比较起来,本系统是B/S结构,学校的考试系统一般为C/S结构,性能方面不如B/S结构,并且C/S接口需要安装客户端,客户端压力很大,我的系统只需要电脑具有浏览器,在同一局域...

计算机视觉--多视几何初步尝试

基础矩阵的原理 K和K’分别是两个相机的参数矩阵。p和p’是X在平面π的坐标表示。所以可以得出 具体计算过程 代码: #!/usr/bin/env python coding: utf-8 from PIL import Image from numpy import * from pylab import * import numpy as np from imp ...

java初学者怎么学习才可以快速入门

java初学者怎么学习才可以快速入门 一、了解JAVA 我们要知道:Java是由Sun Microsystems公司于1995年5月推出的Java面向对象程序设计语言。 Java之父:詹姆斯·高斯林 1.1 java的三个体系 Java SE(Java Platform Standard Edition)。Java SE 以前称为 J2SE。它允许开发和部署在桌面、服务器、嵌入式环境...

猜你喜欢

字段属性之主键&增删改查&自增长&唯一键约束

字段属性之主键&自增长&唯一键约束 主键 主键:primary key 主要的键 一张表中只有一个字段可以使用对应的键,用来唯一的约束该字段里面的数据,不能重复,这种称之为主键 一张表只能最多一个主键 增加主键 SQL操作中有多种方式增加主键大体分为三种 1.在创建表的时候直接在字段之后跟primary key关键字(主键本身不允许为空) 优点:非常直接:缺点:只能使用一个字段作为...

linux下 基于libmad的socket多用户mp3音频在线播放服务器

在众多大神的帮助下,这个在线播放流媒体服务器终于完成啦。。。。 这个mp3流媒体服务器设计的思路是,服务器程序server用多线程实现和多个客户端的通信(这是必然的),然后发送给客户端当前的音频列表公客户端选择,之后根据k客户端的选择给多个客户端传输相应mp3文件的数据,同时,客户端进行实时地音频解码并播放。 关于libmad开源mp3音频解码库的使用,见上一篇博客吧。。。。 在服务器程序这一端,...

Nginx

Nginx Nginx简介: Nginx是一个高性能的http和反向代理服务器,特点是有内存少,并发能力强,事实上Nginx的并发能力确实在同类型网页服务器中表现较好, Nginx用作web服务器:Nginx可以作为静态页面的web服务器,同时还支持CGI语言,但不支持java,java程序只能通过Tomcat配合完成。Nginx专为性能优化而开发,性能是其最重要的考量,实现上非常注重效率,能经受...

SpringCloud Alibaba - Sentinel入门案例(二)(流控规则 | 直接 / 关联 / 链路 / 快速失败 / Warm Up / 排队等待)

SpringCloud Alibaba - Sentinel入门案例(二)(流控规则 | 直接 / 关联 / 链路 / 快速失败 / Warm Up / 排队等待) 回溯 Sentinel 基本概念 正文 环境准备 流控规则介绍 简单介绍 对 阈值类型 / 单机阈值做 测试 流控模式 直接流控模式 关联流控模式 链路流控模式 资源名称的修改 链路模式正文 坑来了,怎么解决? 禁止收敛URL的入口 ...

SQL Server 无法创建索引 因为对象名称和索引名称重复

需求分析:如下图所示,给T_DD_OP1200_Final表的Id字段添加主键,保存时弹出 “T_DD_OP1200_Final”表 - 无法创建索引“PK_T_DD_OP1200_Final”。 因为发现对象名称 ‘dbo.T_DD_OP1200_Final’ 和索引名称 ‘PK_T_DD_OP1200_Fina...