RabbitMQ : 消息确认机制之事务机制

Posted 2021-02-08 refuge

实在没啥好说的.

生产者

    public class Producer
    {
        private const string QueueName = "test_work_queue";
        public static void Send()
        {
            //获取一个连接
            IConnection connection = ConnectionHelper.GetConnection();

            //从连接中获取一个通道
            IModel channel = connection.CreateModel();

            //声明队列
            channel.QueueDeclare(QueueName, false, false, false, null);
            
            //创建一个消息
            string msg = "hello world ";
            try
            {
                //开启事务机制
                //事务机制性能不好,不建议使用.因为需要和服务器发生额外的通信,降低了 RabbitMQ 的吞吐量
                channel.TxSelect();

                //发送消息
                channel.BasicPublish("", QueueName, null, Encoding.Default.GetBytes(msg));

                //提交
                channel.TxCommit();
                Console.WriteLine($"send {msg}");
            }
            catch (Exception e)
            {
                //回滚
                channel.TxRollback();
                Console.WriteLine(e);
            }

            channel.Close();
            connection.Close();
        }
    }

 

消费者

    public class Consumer
    {
        private const string QueueName = "test_work_queue";
        public static void Receive()
        {
            //获取连接
            RabbitMQ.Client.IConnection connection = ConnectionHelper.GetConnection();

            //创建通道
            RabbitMQ.Client.IModel channel = connection.CreateModel();

            //声明队列
            channel.QueueDeclare(QueueName, false, false, false, null);

            //添加消费者
            EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

            //注册事件
            consumer.Received += Consumer_Received;

            //监听队列
            channel.BasicConsume(QueueName, true, "", false, false, null, consumer);
        }

        private static void Consumer_Received(object sender, BasicDeliverEventArgs e)
        {
            byte[] bytes = e.Body;
            string str = Encoding.Default.GetString(bytes);
            Console.WriteLine("consumer : " + str);
        }
    }