上一篇我們說到構建了一個Rabbitmq容器

現在我們說說如何在一個悄悄傳輸訊息到佇列

我們現在設計一個Rabbitmq傳送訊息部分的模組

先設計一個遠端傳送的介面

    public interface IRemoteSend
    {
        void Send<TEntity>(TEntity entity) where TEntity : class;
    }

寫一個Rabbitmq配置實體

    public class RabbitmqConfigura
    {
        public string Host { get
 
; set; }
        public int Port { get; set; }
        public string User { get; set; }
        public string Password { get; set; }
        public string VirtualHost { get; set; }
    }

寫一個實現IRemoteSend的RabbitmqRemoteSend

    public class RabbitmqRemoteSend : IRemoteSend
    {
        private
 
 RabbitmqConfigura Configura { get; }
        public RabbitmqRemoteSend(IOptions<RabbitmqConfigura> options)
        {
            Configura = options.Value;
        }

        public void Send<TEntity>(TEntity entity) where TEntity : class
        {
            throw new NotImplementedException();
        }
    }
 

我們再實現以下Send方法

    public class RabbitmqRemoteSend : IRemoteSend
    {
        public int DelaySend { get; set; }
        private RabbitmqConfigura Configura { get; }
        public RabbitmqRemoteSend(IOptions<RabbitmqConfigura> options)
        {
            Configura = options.Value;
        }

        public void Send<TEntity>(TEntity entity) where TEntity : class
        {
            var factory = new ConnectionFactory
            {
                HostName = Configura.Host,
                Port = Configura.Port,
                UserName = Configura.User,
                Password = Configura.Password,
                VirtualHost = Configura.VirtualHost,

                AutomaticRecoveryEnabled = true,
                NetworkRecoveryInterval = TimeSpan.FromSeconds(30)
            };

            using (var connection = factory.CreateConnection())
            {
                var model = connection.CreateModel();

                var type_name = typeof(TEntity);
                var ExchangeName = type_name + ".exchange";
                var RouteKeyName = type_name + ".input";
                var QueueName = type_name + ".input";

                model.ConfirmSelect();
                model.ExchangeDeclare(ExchangeName, ExchangeType.Direct);
                model.QueueDeclare(QueueName, false, false, false);
                model.QueueBind(QueueName, ExchangeName, RouteKeyName);

                var args = new Dictionary<string, object>();
                args.Add("x-message-ttl", DelaySend);
                args.Add("x-dead-letter-exchange", ExchangeName);
                args.Add("x-dead-letter-routing-key", QueueName);
                model.QueueDeclare(QueueName + ".delay", false, false, false, args);

                var bytes = new byte[];
                var props = model.CreateBasicProperties();

                props.ContentType = "text/plain";
                props.DeliveryMode = 2;
                model.BasicPublish(ExchangeName, RouteKeyName, props, bytes);
                model.WaitForConfirms();
            }
        }
    }

我們需要定義一個序列化的介面做資料編碼

    public interface IFormattor
    {
        byte[] SerializeObject<TEntity>(TEntity entity) where TEntity : class;
    }

寫一個預設實現

    public class JsonFormattor : IFormattor
    {
        public byte[] SerializeObject<TEntity>(TEntity entity) where TEntity : class
        {
            var jsonString = JsonConvert.SerializeObject(entity);
            return Encoding.UTF8.GetBytes(jsonString);
        }
    }

再修改以下RabbitmqRemoteSend

public class RabbitmqRemoteSend : IRemoteSend
    {
        public int DelaySend { get; set; }
        private RabbitmqConfigura Configura { get; }
        private IFormattor Formattor { get; }
        public RabbitmqRemoteSend(IOptions<RabbitmqConfigura> options, IFormattor formattor)
        {
            Configura = options.Value;
            Formattor = formattor;
        }

        public void Send<TEntity>(TEntity entity) where TEntity : class
        {
            var factory = new ConnectionFactory
            {
                HostName = Configura.Host,
                Port = Configura.Port,
                UserName = Configura.User,
                Password = Configura.Password,
                VirtualHost = Configura.VirtualHost,

                AutomaticRecoveryEnabled = true,
                NetworkRecoveryInterval = TimeSpan.FromSeconds(30)
            };

            using (var connection = factory.CreateConnection())
            {
                var model = connection.CreateModel();

                var type_name = typeof(TEntity);
                var ExchangeName = type_name + ".exchange";
                var RouteKeyName = type_name + ".input";
                var QueueName = type_name + ".input";

                model.ConfirmSelect();
                model.ExchangeDeclare(ExchangeName, ExchangeType.Direct);
                model.QueueDeclare(QueueName, false, false, false);
                model.QueueBind(QueueName, ExchangeName, RouteKeyName);

                var args = new Dictionary<string, object>();
                args.Add("x-message-ttl", DelaySend);
                args.Add("x-dead-letter-exchange", ExchangeName);
                args.Add("x-dead-letter-routing-key", QueueName);
                model.QueueDeclare(QueueName + ".delay", false, false, false, args);

                var bytes = Formattor.SerializeObject<TEntity>(entity);
                var props = model.CreateBasicProperties();

                props.ContentType = "text/plain";
                props.DeliveryMode = 2;
                model.BasicPublish(ExchangeName, RouteKeyName, props, bytes);
                model.WaitForConfirms();
            }
        }
    }

我們加入asp.net core測試一下

在ConfigureServices內增加程式碼

            #region Rabbitmq
            services.Configure<RabbitmqConfigura>(p => 
            {
                p.User = "admin";
                p.Password = "123456";
                p.Host = "127.0.0.1";
            });
            services.AddScoped<IFormattor, JsonFormattor>();
            services.AddScoped<IRemoteSend, RabbitmqRemoteSend>();
            #endregion

在Configure增加程式碼

            var scope = app.ApplicationServices.CreateScope();
            var remoteSend = scope.ServiceProvider.GetRequiredService<IRemoteSend>();
            remoteSend.Send(new User()
            {
                Name = "hello",
                Account = "account"
            });

這段程式碼是完全用來測試的

User實體物件

    public class User
    {
        public string Name { get; set; }
        public string Account { get; set; }
    }

我們在看看Rabbitmq內

看看內容