rabbitmq:rpc的实现(代码片段)

private static void MySelfRPCProducer()

    var conn_factory = new ConnectionFactory()HostName = "localhost",UserName = "guest",Password = "guest",Port = 5672;
    using (var conn = conn_factory.CreateConnection())
    
        using (var channel = conn.CreateModel())
        
            IBasicProperties pro = channel.CreateBasicProperties();
            pro.ReplyTo = channel.QueueDeclare().QueueName;//创建应答队列并返回队列名称，这个方法创建的队列exclusive和auto_delete都是true，这样可以确保没有人能窃取信息
            pro.ContentType = "text/plain";
            string corrId = Guid.NewGuid().ToString();
            pro.CorrelationId = corrId;

            channel.BasicPublish("", "rpc_queue", pro, Encoding.UTF8.GetBytes("小黄"));
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (ea, ch) =>
            
                //比较CorrelationId确认是返回的我们的消息
                if (ch.BasicProperties.CorrelationId == corrId)
                
                    //处理返回结果
                    string msg = Encoding.UTF8.GetString(ch.Body);
                    Console.WriteLine(msg);
                
            ;
            string consumer_tag = channel.BasicConsume(pro.ReplyTo, true, consumer);//监听应答队列
            channel.BasicCancel(consumer_tag);
        
    
    Console.ReadLine();

private static void MySelfRPCCousmer()

    var conn_factory = new ConnectionFactory()HostName = "localhost",UserName = "guest",Password = "guest",Port = 5672;
    using (var conn = conn_factory.CreateConnection())
    
        using (var channel = conn.CreateModel())
        
            channel.QueueDeclare("rpc_queue", false, false, false, null);
            var consumer = new EventingBasicConsumer(channel);
            channel.BasicQos(0, 1, false);
            consumer.Received += (ea, ch) =>
            
                string msg = Encoding.UTF8.GetString(ch.Body);
                Console.WriteLine("接收到消息：" + msg);
                //发送处理结果
                channel.BasicPublish("", ch.BasicProperties.ReplyTo, ch.BasicProperties, Encoding.UTF8.GetBytes(msg + "给我回电话了"));
                channel.BasicAck(ch.DeliveryTag, false);
            ;
            string consumer_tag = channel.BasicConsume("rpc_queue", false, consumer);
            Console.ReadLine();//这里先停止运行下面的代码，因为需要持续监听，信道断开就监听不了了
            channel.BasicCancel(consumer_tag);
        
    

private static void RabbitMQRPCProducer()

    var conn_factory = new ConnectionFactory()  HostName = "localhost", UserName = "guest", Password = "guest", Port = 5672 ;
    using (var conn = conn_factory.CreateConnection())
    
        using (var channel = conn.CreateModel())
        
            //创建client的rpc
            SimpleRpcClient client = new SimpleRpcClient(channel, new PublicationAddress(exchangeType: ExchangeType.Direct, exchangeName: string.Empty, routingKey: "rpc_queue"));
            bool flag = true;
            var sendmsg = "";
            while (flag)
            
                Console.WriteLine("请输入要发送的消息");
                sendmsg = Console.ReadLine();
                if (string.IsNullOrWhiteSpace(sendmsg))
                
                    Console.Write("请输入消息");
                    continue;
                
                var msg = client.Call(Encoding.UTF8.GetBytes(sendmsg));
                Console.WriteLine(Encoding.UTF8.GetString(msg));
            
            Console.ReadKey();
        
    

private static void RabbitMQRPCCousmer()


    var conn_factory = new ConnectionFactory()  HostName = "localhost", UserName = "guest", Password = "guest", Port = 5672 ;
    using (var conn = conn_factory.CreateConnection())
    
        //创建返回一个新的频道
        using (var channel = conn.CreateModel())
        
            channel.QueueDeclare("rpc_queue", false, false, false, null);//创建一个rpc queue
            SimpleRpcServer rpc = new MySimpleRpcServer(new Subscription(channel, "rpc_queue"));
            Console.WriteLine("服务端启动成功");
            rpc.MainLoop(); Console.ReadKey();
        
    

class MySimpleRpcServer : SimpleRpcServer

    public MySimpleRpcServer(Subscription subscription) : base(subscription)
    
    
    /// <summary>
    /// 执行完成后进行回调
    /// </summary>
    public override byte[] HandleSimpleCall(bool isRedelivered, IBasicProperties requestProperties, byte[] body, out IBasicProperties replyProperties)
    
        replyProperties = null;
        return Encoding.UTF8.GetBytes($"给Encoding.UTF8.GetString(body)发送短信成功");