乘风原创程序

  • c# rabbitmq 简单收发消息的示例代码
  • 2020/8/19 9:49:30
  • 发布消息:(生产者)

       /// <summary>
        /// 发送消息
        /// </summary>
        /// <param name="queue">队列名</param>
        /// <param name="message">消息内容</param>
        private static void PublishInfo(string queue, string message)
        {
          try
          {
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";//RabbitMQ服务在本地运行
            factory.UserName = "guest";//用户名
            factory.Password = "guest";//密码
    
            using (var connection = factory.CreateConnection())
            {
              using (var channel = connection.CreateModel())
              {
                bool durable = true;//将消息标记为持久性
                channel.QueueDeclare(queue, durable, false, false, null);
    
                IBasicProperties properties = channel.CreateBasicProperties();
                properties.DeliveryMode = 2;
    
                var body = Encoding.UTF8.GetBytes(message);
                channel.BasicPublish("", queue, properties, body); //开始传递
    
              }
            }
          }
          catch (System.Exception ex)
          {
            Console.WriteLine(ex.ToString());
          }
          Console.ReadLine();
        }

    使用:

    PublishInfo("test", "222222222222");

     点击test进入队列内部:

    消费消息:

    方法一:旧版可用

     private static void ConsumeInfo(string queue)
        {
          try
          {
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";//RabbitMQ服务在本地运行
            factory.UserName = "guest";//用户名
            factory.Password = "guest";//密码
    
            using (var connection = factory.CreateConnection())
            {
              using (var channel = connection.CreateModel())
              {
                //bool durable = true;//将消息标记为持久性
                //channel.QueueDeclare("sljcgx:finish-queue", durable, false, false, null);//声明队列,队列不存在则创建该队列
    
                //设置prefetchCount : 1来告知RabbitMQ,在未收到消费端的消息确认时,不再分发消息,也就确保了当消费端处于忙碌状态时,不再分配任务。
                channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
                var consumer = new QueueingBasicConsumer(channel);
               
                //false为手动确认,获取消息后需要添加 channel.BasicAck(ea.DeliveryTag, false)确认
                //true则为自动确认,不需要这行代码
                channel.BasicConsume(queue, false, consumer);
    
                while (true)
                {
                  try
                  {
                    var ea = (BasicDeliverEventArgs)consumer.Queue.Dequeue();//阻塞函数,获取队列中的消息
                    var bodyReceive = ea.Body;
                    string message = Encoding.UTF8.GetString(bodyReceive);
                    Console.WriteLine(message);
                    channel.BasicAck(ea.DeliveryTag, false);// 消息确认信号,确认后该消息将从队列里移除
                  }
                  catch (System.Exception ex)
                  {
                    
                    Console.WriteLine(ex.Message);
                  }
                }
              }
    
            }
          }
          catch (System.Exception ex)
          {
            Console.WriteLine(ex.ToString());
          }
          Console.ReadLine();
        }

    方法二:(通用)

       private static void ConsumeInfo2(string queue)
        {
          try
          {
            var factory = new ConnectionFactory();
            factory.HostName = "localhost";//RabbitMQ服务在本地运行
            factory.UserName = "guest";//用户名
            factory.Password = "guest";//密码
            var connection = factory.CreateConnection();
            var channel = connection.CreateModel();
            channel.QueueDeclare(queue, true, false, false, null);//声明队列,队列不存在则创建该队列
            channel.BasicQos(0, 1, false);//公平分发、同一时间只处理一个消息。
    
            var consumer = new EventingBasicConsumer(channel); //消费者(指定消息通道)该事件在接收到消息时触发
            consumer.Received += (sender, e) =>
            {
              byte[] body = e.Body.ToArray();  //消息字节数组
              string message = Encoding.UTF8.GetString(body); //消息内容
    
              Console.WriteLine(message);
              channel.BasicAck(e.DeliveryTag, false);    //手工确认
    
            };
            channel.BasicConsume(queue, false, consumer);  //消费消息(在当前通道中监听 queue 队列,并进行消费)
            Console.ReadLine();
            connection.Close();
            channel.Close();
          }
          catch (System.Exception ex)
          {
            Console.WriteLine(ex.ToString());
          }
        }