# RabbitMQ_Demo

**Repository Path**: ldhnet/RabbitMQ

## Basic Information

- **Project Name**: RabbitMQ_Demo
- **Description**: 基于DotNet 6 实现的RabbitMQ demo 开箱即用，MQ 解耦，异步，削峰等实用利器。   从https://github.com/ldhnet迁移到gitee平台
- **Primary Language**: Unknown
- **License**: MIT
- **Default Branch**: master
- **Homepage**: https://github.com/ldhnet
- **GVP Project**: No

## Statistics

- **Stars**: 3
- **Forks**: 0
- **Created**: 2018-03-15
- **Last Updated**: 2024-06-28

## Categories & Tags

**Categories**: Uncategorized

**Tags**: dotNET

## README

<h4 align="center">RabbitMQDemo</h4>
<p align="center">
	<a href="https://gitee.com/ldhnet/RepositoryUnitOfWork/stargazers"><img src="https://gitee.com/ldhnet/RepositoryUnitOfWork/badge/star.svg?theme=dark"></a>
</p>


        public RabbitMQManager(IOptionsMonitor<RabbitMQOptions> options)
        {
            _options = options.CurrentValue;

            var factory = new ConnectionFactory()
            {
                HostName = _options.HostName,
                UserName = _options.UserName,
                Password = _options.Password
            };
            factory.AutomaticRecoveryEnabled = true;   //设置端口后 自动恢复连接属性即可
            //创建连接
            _connection = factory.CreateConnection();           
        }

        public void Consume<T>(string queueName, string routeKey, Action<T> action) where T : class
        {
            //创建通道
            var channel = _connection.CreateModel();

            //声明一个队列 (durable=true 持久化消息）
            channel.QueueDeclare(queueName, true, false, false, null);

            if (!string.IsNullOrEmpty(_options.ExchangeName))
            {
                channel.ExchangeDeclare(_options.ExchangeName, _options.ExchangeType, true, false, null);
                //将队列绑定到交换机
                channel.QueueBind(queueName, _options.ExchangeName, routeKey, null);
            }

            channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);            

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                var obj = JsonConvert.DeserializeObject<T>(message);

                try
                {                    
                    action?.Invoke(obj);                    
                }
                catch (Exception ex)
                {
                    throw ex;
                }
                channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);                                   
            };

            channel.BasicConsume(queue: queueName,
                                 autoAck: false,
                                 consumer: consumer);            
        }

        public void Publish<T>(T message, string queueName, string routeKey)
        {
            if (message == null || string.IsNullOrWhiteSpace(queueName))
                return;            

            //创建通道
            var channel = _connection.CreateModel();

            //声明一个队列 (durable=true 持久化消息）
            channel.QueueDeclare(queueName, true, false, false, null);

            if (!string.IsNullOrEmpty(_options.ExchangeName))
            {
                channel.ExchangeDeclare(_options.ExchangeName, _options.ExchangeType, true, false, null);
                //将队列绑定到交换机
                channel.QueueBind(queueName, _options.ExchangeName, routeKey, null);
            }

            var sendBytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message));

            var properties = channel.CreateBasicProperties();
            properties.Persistent = true;

            channel.BasicPublish(_options.ExchangeName, routeKey, properties, sendBytes);
        }