channel.BasicReturn += Channel_BasicReturn;
channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: null, body: body);

private static void Channel_BasicReturn(object sender, RabbitMQ.Client.Events.BasicReturnEventArgs e)
{
    Console.WriteLine("Channel_BasicReturn");
}

channel.ConfirmSelect();

channel.BasicPublish("amq.direct", routingKey: "MyRoutKey1",mandatory:true, basicProperties: null, body: body);

if (channel.WaitForConfirms())
{
    Console.WriteLine("普通发送方确认模式");
}

channel.ConfirmSelect();

channel.BasicPublish("amq.direct", routingKey: "MyRoutKey",mandatory:true, basicProperties: null, body: body);
channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: null, body: body);
channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: null, body: body);

channel.WaitForConfirmsOrDie();
Console.WriteLine("普通发送方确认模式");

try
{
    //声明事物
    channel.TxSelect();
    channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: null, body: body);
    channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: null, body: body);
    channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: null, body: body);
    //提交事物
    channel.TxCommit();
}
catch (Exception)
{
    //回滚
    channel.TxRollback();

}

static void Main(string[] args)
        {


            ConnectionFactory factory = new ConnectionFactory();
            factory.HostName = "192.168.140.161";
            factory.Port = 5672;
            factory.UserName = "admin";
            factory.Password = "admin";
            factory.VirtualHost = "TestVHost";
            //创建connetion
            using (var connetion = factory.CreateConnection())
            {
                connetion.CallbackException += Connetion_CallbackException;
                connetion.RecoverySucceeded += Connetion_RecoverySucceeded;
                connetion.ConnectionRecoveryError += Connetion_ConnectionRecoveryError;
                connetion.ConnectionBlocked += Connetion_ConnectionBlocked;
                connetion.ConnectionUnblocked += Connetion_ConnectionUnblocked;
                //连接关闭的时候
                connetion.ConnectionShutdown += Connetion_ConnectionShutdown;



                //创建channel
                using (var channel = connetion.CreateModel())
                {

                    //消息会在何时被 confirm？
                    //The broker will confirm messages once:
                    //broker 将在下面的情况中对消息进行 confirm ：
                    //it decides a message will not be routed to queues
                    //(if the mandatory flag is set then the basic.return is sent first) or
                    //broker 发现当前消息无法被路由到指定的 queues 中（如果设置了 mandatory 属性，则 broker 会先发送 basic.return）
                    //a transient message has reached all its queues(and mirrors) or
                    //非持久属性的消息到达了其所应该到达的所有 queue 中（和镜像 queue 中）
                    //a persistent message has reached all its queues(and mirrors) and been persisted to disk(and fsynced) or
                    //持久消息到达了其所应该到达的所有 queue 中（和镜像 queue 中），并被持久化到了磁盘（被 fsync）
                    //a persistent message has been consumed(and if necessary acknowledged) from all its queues
                    //持久消息从其所在的所有 queue 中被 consume 了（如果必要则会被 acknowledge）


                    //broker 发现当前消息无法被路由到指定的 queues 中（如果设置了 mandatory 属性，则 broker 会先发送 basic.return）
                    channel.BasicReturn += Channel_BasicReturn;

                    //(可以不声明)如果不声明交换机 ，那么就使用默认的交换机  （每一个vhost都会有一个默认交换机）
                    //channel.ExchangeDeclare("amq.direct", ExchangeType.Direct,true);

                    //创建一个队列  bool durable(持久化), bool exclusive（专有的）, bool autoDelete（自动删除）
                    //channel.QueueDeclare("TestQueue", true, false, false, null);
                    //不做绑定的话，使用默认的交换机。
                    //channel.QueueBind("TestQueue", "amq.direct", "MyRoutKey", null);

                    //发布消息
                    var body = Encoding.UTF8.GetBytes("西伯利亚的狼");


                    channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: null, body: body);
                }

                Console.WriteLine("Hello World!");
                Console.ReadKey();
            }
        }

private static void Channel_BasicReturn(object sender, RabbitMQ.Client.Events.BasicReturnEventArgs e)
        {
            Console.WriteLine("Channel_BasicReturn");
        }

private static void Connetion_ConnectionShutdown(object sender, ShutdownEventArgs e)
        {
            Console.WriteLine("Connetion_ConnectionShutdown");
        }

private static void Connetion_ConnectionUnblocked(object sender, EventArgs e)
        {
            Console.WriteLine("Connetion_ConnectionUnblocked");
        }

private static void Connetion_ConnectionBlocked(object sender, RabbitMQ.Client.Events.ConnectionBlockedEventArgs e)
        {
            Console.WriteLine("Connetion_ConnectionBlocked");
        }

private static void Connetion_ConnectionRecoveryError(object sender, RabbitMQ.Client.Events.ConnectionRecoveryErrorEventArgs e)
        {
            Console.WriteLine("Connetion_ConnectionRecoveryError");
        }

private static void Connetion_RecoverySucceeded(object sender, EventArgs e)
{
    Console.WriteLine("Connetion_RecoverySucceeded");
}

private static void Connetion_CallbackException(object sender, RabbitMQ.Client.Events.CallbackExceptionEventArgs e)
{
    Console.WriteLine("Connetion_CallbackException");
}

//发布消息
var body = Encoding.UTF8.GetBytes("西伯利亚的狼");
BasicProperties pro = new BasicProperties();
pro.DeliveryMode = 2;

channel.BasicPublish("amq.direct", routingKey: "MyRoutKey", mandatory: true, basicProperties: pro, body: body);

ConnectionFactory factory = new ConnectionFactory();
factory.HostName = "192.168.140.161";
factory.Port = 5672;
factory.UserName = "admin";
factory.Password = "admin";
factory.VirtualHost = "TestVHost";
//创建connetion
using (var connetion = factory.CreateConnection())
{

    using (var channel = connetion.CreateModel())
    {
        //构造消费者实例
        var consumer = new EventingBasicConsumer(channel);
        //绑定消息接收后的事件委托
        consumer.Received += (model, ea) =>
        {
            var message = Encoding.UTF8.GetString(ea.Body);
            Console.WriteLine(" [x] Received {0}", message);
            Thread.Sleep(6000);//模拟耗时
            Console.WriteLine(" [x] Done");
            // 主要改动的是将 autoAck:true修改为autoAck:fasle，以及在消息处理完毕后手动调用BasicAck方法进行手动消息确认。
            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
        };
        //启动消费者
        channel.BasicConsume(queue: "TestQueue", autoAck: false, consumer: consumer);
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();

    }
}

using (var connetion = factory.CreateConnection())
{

    using (var channel = connetion.CreateModel())
    {
        //构造消费者实例
        var consumer = new EventingBasicConsumer(channel);
        //绑定消息接收后的事件委托
        consumer.Received += (model, ea) =>
        {
            var message = Encoding.UTF8.GetString(ea.Body);
            Console.WriteLine(" [x] Received {0}", message);
            Thread.Sleep(6000);//模拟耗时
            Console.WriteLine(" [x] Done");
        };
        //启动消费者
        channel.BasicConsume(queue: "TestQueue", autoAck: true, consumer: consumer);
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();

    }
}

//申明唯一guid用来标识此次发送的远程调用请求
 var correlationId = Guid.NewGuid().ToString();
 //申明需要监听的回调队列
 var replyQueue = channel.QueueDeclare().QueueName;
 var properties = channel.CreateBasicProperties();
 properties.ReplyTo = replyQueue;//指定回调队列
 properties.CorrelationId = correlationId;//指定消息唯一标识
 string number = args.Length > 0 ? args[0] : "30";
 var body = Encoding.UTF8.GetBytes(number);
 //发布消息
 channel.BasicPublish(exchange: "", routingKey: "rpc_queue", basicProperties: properties, body: body);
 Console.WriteLine($"[*] Request fib({number})");
 // //创建消费者用于处理消息回调（远程调用返回结果）
 var callbackConsumer = new EventingBasicConsumer(channel);
 channel.BasicConsume(queue: replyQueue, autoAck: true, consumer: callbackConsumer);
 callbackConsumer.Received += (model, ea) =>
 {
      //仅当消息回调的ID与发送的ID一致时，说明远程调用结果正确返回。
     if (ea.BasicProperties.CorrelationId == correlationId)
     {
         var responseMsg = $"Get Response: {Encoding.UTF8.GetString(ea.Body)}";
         Console.WriteLine($"[x]: {responseMsg}");
     }
 };

//申明队列接收远程调用请求
channel.QueueDeclare(queue: "rpc_queue", durable: false,
    exclusive: false, autoDelete: false, arguments: null);
var consumer = new EventingBasicConsumer(channel);
Console.WriteLine("[*] Waiting for message.");
//请求处理逻辑
consumer.Received += (model, ea) =>
{
    var message = Encoding.UTF8.GetString(ea.Body);
    int n = int.Parse(message);
    Console.WriteLine($"Receive request of Fib({n})");
    int result = Fib(n);
    //从请求的参数中获取请求的唯一标识，在消息回传时同样绑定
    var properties = ea.BasicProperties;
    var replyProerties = channel.CreateBasicProperties();
    replyProerties.CorrelationId = properties.CorrelationId;
    //将远程调用结果发送到客户端监听的队列上
    channel.BasicPublish(exchange: "", routingKey: properties.ReplyTo,
        basicProperties: replyProerties, body: Encoding.UTF8.GetBytes(result.ToString()));
    //手动发回消息确认
    channel.BasicAck(ea.DeliveryTag, false);
    Console.WriteLine($"Return result: Fib({n})= {result}");
};
channel.BasicConsume(queue: "rpc_queue", autoAck: false, consumer: consumer);

    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "localhost", VirtualHost = "OrderQueue", UserName = "zhangweizhong", Password = "weizhong1988", Port = 5672 };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(queue: "rpc_queue",
                                     durable: false,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);
                channel.BasicQos(0, 1, false);
                var consumer = new QueueingBasicConsumer(channel);
                channel.BasicConsume(queue: "rpc_queue",
                                     noAck: false,
                                     consumer: consumer);
                Console.WriteLine(" [x] Awaiting RPC requests");

                while (true)
                {
                    string response = null;
                    var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();

                    var body = ea.Body;
                    var props = ea.BasicProperties;
                    var replyProps = channel.CreateBasicProperties();
                    replyProps.CorrelationId = props.CorrelationId;

                    try
                    {
                        var message = Encoding.UTF8.GetString(body);
                        int n = int.Parse(message);
                        Console.WriteLine(" [.] fib({0})", message);
                        response = fib(n).ToString();
                    }
                    catch (Exception e)
                    {
                        Console.WriteLine(" [.] " + e.Message);
                        response = "";
                    }
                    finally
                    {
                        var responseBytes = Encoding.UTF8.GetBytes(response);
                        channel.BasicPublish(exchange: "",
                                             routingKey: props.ReplyTo,
                                             basicProperties: replyProps,
                                             body: responseBytes);
                        channel.BasicAck(deliveryTag: ea.DeliveryTag,
                                         multiple: false);
                    }
                }
            }
        }

        /// <summary>
        /// Assumes only valid positive integer input.
        /// Don't expect this one to work for big numbers,
        /// and it's probably the slowest recursive implementation possible.
        /// </summary>
        private static int fib(int n)
        {
            if (n == 0 || n == 1)
            {
                return n;
            }

            Thread.Sleep(1000 * 10);

            return n;
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            for (int i = 0; i < 10; i++)
            {
                Stopwatch watch = new Stopwatch();

                watch.Start();

                var rpcClient = new RPCClient();

                Console.WriteLine(string.Format(" [x] Requesting fib({0})", i));

                var response = rpcClient.Call(i.ToString());

                Console.WriteLine(" [.] Got '{0}'", response);

                rpcClient.Close();

                watch.Stop();

                Console.WriteLine(string.Format(" [x] Requesting complete {0} ,cost {1} ms", i, watch.Elapsed.TotalMilliseconds));
            }

            Console.WriteLine(" complete!!!! ");


            Console.ReadLine();
        }
    }

    class RPCClient
    {
        private IConnection connection;
        private IModel channel;
        private string replyQueueName;
        private QueueingBasicConsumer consumer;

        public RPCClient()
        {
            var factory = new ConnectionFactory() { HostName = "localhost", VirtualHost = "OrderQueue", UserName = "zhangweizhong", Password = "weizhong1988", Port = 5672 };
            connection = factory.CreateConnection();
            channel = connection.CreateModel();
            replyQueueName = channel.QueueDeclare().QueueName;
            consumer = new QueueingBasicConsumer(channel);
            channel.BasicConsume(queue: replyQueueName,
                                 noAck: true,
                                 consumer: consumer);
        }

        public string Call(string message)
        {
            var corrId = Guid.NewGuid().ToString();
            var props = channel.CreateBasicProperties();
            props.ReplyTo = replyQueueName;
            props.CorrelationId = corrId;

            var messageBytes = Encoding.UTF8.GetBytes(message);
            channel.BasicPublish(exchange: "",
                                 routingKey: "rpc_queue",
                                 basicProperties: props,
                                 body: messageBytes);

            while (true)
            {
                var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
                if (ea.BasicProperties.CorrelationId == corrId)
                {
                    return Encoding.UTF8.GetString(ea.Body);
                }
            }
        }

        public void Close()
        {
            connection.Close();
        }
    }

　　　　 /// <summary>
        /// 生产者
        /// </summary>
        /// <param name="change"></param>
        private static void ProducerMessage(MyMessage msg)
        {
            var advancedBus = CreateAdvancedBus();

            if (advancedBus.IsConnected)
            {
                var exchange = advancedBus.ExchangeDeclare("user", ExchangeType.Fanout);

                advancedBus.Publish(exchange, "", false, new Message<MyMessage>(msg));
            }
            else
            {
                Console.WriteLine("Can't connect");
            }

        }

        /// <summary>
        /// 消费者
        /// </summary>
        private static void ConsumeMessage()
        {
            var advancedBus = CreateAdvancedBus();
            var exchange = advancedBus.ExchangeDeclare("user", ExchangeType.Fanout);

            var queue = advancedBus.QueueDeclare("user.notice.wangwu");
            advancedBus.Bind(exchange, queue, "user.notice.wangwu");
            advancedBus.Consume(queue, registration =>
            {
                registration.Add<MyMessage>((message, info) => { Console.WriteLine("Body: {0}", message.Body); });
            });
        }

public static IAdvancedBus CreateAdvancedBus()
{
    // 消息服务器连接字符串
    string connString = "host=dev.corp.wingoht.com:5672;virtualHost=cd;username=ishowfun;password=123456";
    if (connString == null || connString == string.Empty)
    {
        throw new Exception("messageserver connection string is missing or empty");
    }

    return RabbitHutch.CreateBus(connString).Advanced;
}

     /// <summary>
        /// 生产者
        /// </summary>
        /// <param name="change"></param>
        private static void ProducerMessage(MyMessage msg)
        {
            var advancedBus = CreateAdvancedBus();

            if (advancedBus.IsConnected)
            {
                var queue = advancedBus.QueueDeclare("user.notice.zhangsan");

                advancedBus.Publish(Exchange.GetDefault(), queue.Name, false, new Message<MyMessage>(msg));
            }
            else
            {
                Console.WriteLine("Can't connect");
            }

        }

        /// <summary>
        /// 消费者
        /// </summary>
        private static void ConsumeMessage()
        {
            var advancedBus = CreateAdvancedBus();

            var exchange = advancedBus.ExchangeDeclare("user", ExchangeType.Direct);

            var queue = advancedBus.QueueDeclare("user.notice.lisi");

            advancedBus.Bind(exchange, queue, "user.notice.lisi");

            advancedBus.Consume(queue, registration =>
            {
                registration.Add<MyMessage>((message, info) =>
                {
                    Console.WriteLine("Body: {0}", message.Body);
                });
            });
        }

　　　　　/// <summary>
        /// 生产者
        /// </summary>
        /// <param name="change"></param>
        private static void ProducerMessage(MyMessage msg)
        {
            //// 创建消息bus
            IBus bus = CreateBus();

            try
            {
                bus.Publish(msg, x => x.WithTopic(msg.MessageRouter));
            }
            catch (EasyNetQException ex)
            {
                //处理连接消息服务器异常 
            }

            bus.Dispose();//与数据库connection类似，使用后记得销毁bus对象
        }

        /// <summary>
        /// 消费者
        /// </summary>
        private static void ConsumeMessage(MyMessage msg)
        {
            //// 创建消息bus
            IBus bus = CreateBus();

            try
            {
                bus.Subscribe<MyMessage>(msg.MessageRouter, message => Console.WriteLine(msg.MessageBody), x => x.WithTopic("user.notice.#"));
            }
            catch (EasyNetQException ex)
            {
                //处理连接消息服务器异常 
            }
        }