c#開源訊息佇列中介軟體EQueue 教程

阿新 • • 發佈：2019-01-20

一、簡介

EQueue是一個參照RocketMQ實現的開源訊息佇列中介軟體,相容Mono，具體可以參看作者的文章《分享一個c#寫的開源分散式訊息佇列equeue》。專案開源地址：https://github.com/tangxuehua/equeue，專案中包含了佇列的全部原始碼以及如何使用的示例。二、安裝EQueue Producer、Consumer、Broker支援分散式部署，安裝EQueue需要.NET 4, Visual Studio 2010/2012/2013. 目前EQueue是個類庫，需要自己實現Broker的宿主，可以參照QuickStart，建立一個QuickStart.BrokerServer專案，通過Visual Studio的Nuget 查詢equeue

using System;
using System.Text;
using ECommon.Autofac;
using ECommon.Configurations;
using ECommon.JsonNet;
using ECommon.Log4Net;
using EQueue.Broker;
using EQueue.Configurations;
using EQueue.Protocols;

namespace QuickStart.BrokerServer {
    class Program {
        static void Main(string[] args) {
            InitializeEQueue();
            var setting = new BrokerSetting();
            setting.NotifyWhenMessageArrived = false;
            setting.DeleteMessageInterval = 1000;
            new BrokerController(setting).Initialize().Start();
            Console.ReadLine();
        }

        static void InitializeEQueue() {
            Configuration.Create().UseAutofac().RegisterCommonComponents().UseLog4Net().UseJsonNet().RegisterEQueueComponents();
        }
    }
}

InitializeEQueue方法初始化EQueue的環境，使用了Autofac作為IOC容器，使用log4Net記錄日誌, 我們看一下RegisterEQueueComponents方法：

public static class ConfigurationExtensions {
    public static Configuration RegisterEQueueComponents(this Configuration configuration) {
        configuration.SetDefault < IAllocateMessageQueueStrategy,
        AverageAllocateMessageQueueStrategy > ();
        configuration.SetDefault < IQueueSelector,
        QueueHashSelector > ();
        configuration.SetDefault < ILocalOffsetStore,
        DefaultLocalOffsetStore > ();
        configuration.SetDefault < IMessageStore,
        InMemoryMessageStore > ();
        configuration.SetDefault < IMessageService,
        MessageService > ();
        configuration.SetDefault < IOffsetManager,
        InMemoryOffsetManager > ();
        return configuration;
    }
}

程式碼中涉及到6個元件:
IAllocateMessageQueueStrategy
IQueueSelector
ILocalOffsetStore
IMessageStore
IMessageService
IOffsetManager
DeleteMessageInterval 這個屬性是用來設定equeue的定時刪除間隔，單位為毫秒,預設值是一個小時。另外還有ProducerSocketSetting 和 ConsumerSocketSetting 分別用於設定Producer連線Broker和Consumer連線Broker的IP和埠，預設埠是5000和5001。

public class BrokerSetting {
    public SocketSetting ProducerSocketSetting {
        get;
        set;
    }
    public SocketSetting ConsumerSocketSetting {
        get;
        set;
    }
    public bool NotifyWhenMessageArrived {
        get;
        set;
    }
    public int DeleteMessageInterval {
        get;
        set;
    }

    public BrokerSetting() {
        ProducerSocketSetting = new SocketSetting {
            Address = SocketUtils.GetLocalIPV4().ToString(),
            Port = 5000,
            Backlog = 5000
        };
        ConsumerSocketSetting = new SocketSetting {
            Address = SocketUtils.GetLocalIPV4().ToString(),
            Port = 5001,
            Backlog = 5000
        };
        NotifyWhenMessageArrived = true;
        DeleteMessageInterval = 1000 * 60 * 60;
    }
}

執行專案，如果顯示下面類似內容，說明Broker啟動成功：
2014-03-23 20:10:30,255 INFO BrokerController - Broker started, producer:[169.254.80.80:5000], consumer:[169.254.80.80:5001] 三、在Visual Studio中開發測試 1.建立一個VS專案 QuickStart.ProducerClient，通過Nuget引用EQueue，編寫下面Producer程式碼

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using ECommon.Autofac;
using ECommon.Configurations;
using ECommon.IoC;
using ECommon.JsonNet;
using ECommon.Log4Net;
using ECommon.Scheduling;
using EQueue.Clients.Producers;
using EQueue.Configurations;
using EQueue.Protocols;

namespace QuickStart.ProducerClient {
    class Program {
        static void Main(string[] args) {
            InitializeEQueue();

            var scheduleService = ObjectContainer.Resolve < IScheduleService > ();
            var producer = new Producer().Start();
            var total = 1000;
            var parallelCount = 10;
            var finished = 0;
            var messageIndex = 0;
            var watch = Stopwatch.StartNew();

            var action = new Action(() = >{
                for (var index = 1; index <= total; index++) {
                    var message = "message" + Interlocked.Increment(ref messageIndex);
                    producer.SendAsync(new Message("SampleTopic", Encoding.UTF8.GetBytes(message)), index.ToString()).ContinueWith(sendTask = >{
                        var finishedCount = Interlocked.Increment(ref finished);
                        if (finishedCount % 1000 == 0) {
                            Console.WriteLine(string.Format("Sent {0} messages, time spent:{1}", finishedCount, watch.ElapsedMilliseconds));
                        }
                    });
                }
            });

            var actions = new List < Action > ();
            for (var index = 0; index < parallelCount; index++) {
                actions.Add(action);
            }

            Parallel.Invoke(actions.ToArray());

            Console.ReadLine();
        }

        static void InitializeEQueue() {
            Configuration.Create().UseAutofac().RegisterCommonComponents().UseLog4Net().UseJsonNet().RegisterEQueueComponents();
        }
    }
}

Producer物件在使用之前必須要呼叫Start初始化，初始化一次即可，注意：切記不可以在每次傳送訊息時，都呼叫Start方法。Producer 預設連線本機的5000埠，可以通過ProducerSetting 進行設定，可以參看下面的程式碼：

public class ProducerSetting {
    public string BrokerAddress {
        get;
        set;
    }
    public int BrokerPort {
        get;
        set;
    }
    public int SendMessageTimeoutMilliseconds {
        get;
        set;
    }
    public int UpdateTopicQueueCountInterval {
        get;
        set;
    }

    public ProducerSetting() {
        BrokerAddress = SocketUtils.GetLocalIPV4().ToString();
        BrokerPort = 5000;
        SendMessageTimeoutMilliseconds = 1000 * 10;
        UpdateTopicQueueCountInterval = 1000 * 5;
    }

2、建立一個VS專案 QuickStart.ConsumerClient，通過Nuget引用EQueue，編寫下面Consumer程式碼

using System;
using System.Linq;
using System.Text;
using System.Threading;
using ECommon.Autofac;
using ECommon.Configurations;
using ECommon.IoC;
using ECommon.JsonNet;
using ECommon.Log4Net;
using ECommon.Scheduling;
using EQueue.Broker;
using EQueue.Clients.Consumers;
using EQueue.Configurations;
using EQueue.Protocols;

namespace QuickStart.ConsumerClient {
    class Program {
        static void Main(string[] args) {
            InitializeEQueue();

            var messageHandler = new MessageHandler();
            var consumer1 = new Consumer("Consumer1", "group1").Subscribe("SampleTopic").Start(messageHandler);
            var consumer2 = new Consumer("Consumer2", "group1").Subscribe("SampleTopic").Start(messageHandler);
            var consumer3 = new Consumer("Consumer3", "group1").Subscribe("SampleTopic").Start(messageHandler);
            var consumer4 = new Consumer("Consumer4", "group1").Subscribe("SampleTopic").Start(messageHandler);

            Console.WriteLine("Start consumer load balance, please wait for a moment.");
            var scheduleService = ObjectContainer.Resolve < IScheduleService > ();
            var waitHandle = new ManualResetEvent(false);
            var taskId = scheduleService.ScheduleTask(() = >{
                var c1AllocatedQueueIds = consumer1.GetCurrentQueues().Select(x = >x.QueueId);
                var c2AllocatedQueueIds = consumer2.GetCurrentQueues().Select(x = >x.QueueId);
                var c3AllocatedQueueIds = consumer3.GetCurrentQueues().Select(x = >x.QueueId);
                var c4AllocatedQueueIds = consumer4.GetCurrentQueues().Select(x = >x.QueueId);
                if (c1AllocatedQueueIds.Count() == 1 && c2AllocatedQueueIds.Count() == 1 && c3AllocatedQueueIds.Count() == 1 && c4AllocatedQueueIds.Count() == 1) {
                    Console.WriteLine(string.Format("Consumer load balance finished. Queue allocation result: c1:{0}, c2:{1}, c3:{2}, c4:{3}", string.Join(",", c1AllocatedQueueIds), string.Join(",", c2AllocatedQueueIds), string.Join(",", c3AllocatedQueueIds), string.Join(",", c4AllocatedQueueIds)));
                    waitHandle.Set();
                }
            },
            1000, 1000);

            waitHandle.WaitOne();
            scheduleService.ShutdownTask(taskId);

            Console.ReadLine();
        }

        static void InitializeEQueue() {
            Configuration.Create().UseAutofac().RegisterCommonComponents().UseLog4Net().UseJsonNet().RegisterEQueueComponents();
        }
    }

    class MessageHandler: IMessageHandler {
        private int _handledCount;

        public void Handle(QueueMessage message, IMessageContext context) {
            var count = Interlocked.Increment(ref _handledCount);
            if (count % 1000 == 0) {
                Console.WriteLine("Total handled {0} messages.", count);
            }
            context.OnMessageHandled(message);
        }
    }
}

使用方式給使用者感覺是訊息從EQueue伺服器推到了應用客戶端。但是實際Consumer內部是使用長輪詢Pull方式從EQueue伺服器拉訊息，然後再回呼叫戶Listener方法。Consumer預設連線本機的5001埠，可以通過ConsumerSetting 進行設定，可以參看下面的程式碼：

public class ConsumerSetting {
    public string BrokerAddress {
        get;
        set;
    }
    public int BrokerPort {
        get;
        set;
    }
    public int RebalanceInterval {
        get;
        set;
    }
    public int UpdateTopicQueueCountInterval {
        get;
        set;
    }
    public int HeartbeatBrokerInterval {
        get;
        set;
    }
    public int PersistConsumerOffsetInterval {
        get;
        set;
    }
    public PullRequestSetting PullRequestSetting {
        get;
        set;
    }
    public MessageModel MessageModel {
        get;
        set;
    }
    public MessageHandleMode MessageHandleMode {
        get;
        set;
    }

    public ConsumerSetting() {
        BrokerAddress = SocketUtils.GetLocalIPV4().ToString();
        BrokerPort = 5001;
        RebalanceInterval = 1000 * 5;
        HeartbeatBrokerInterval = 1000 * 5;
        UpdateTopicQueueCountInterval = 1000 * 5;
        PersistConsumerOffsetInterval = 1000 * 5;
        PullRequestSetting = new PullRequestSetting();
        MessageModel = MessageModel.Clustering;
        MessageHandleMode = MessageHandleMode.Parallel;
    }

c#開源訊息佇列中介軟體EQueue 教程

c#開源訊息佇列中介軟體EQueue 教程

訊息佇列中介軟體？？

Delayer 基於 Redis 的延遲訊息佇列中介軟體

訊息佇列中介軟體（一）介紹

訊息佇列中介軟體（二）使用 ActiveMQ

訊息佇列中介軟體（三）Kafka 入門指南

什麼是訊息佇列中介軟體詳解

常用的訊息佇列中介軟體mq對比

訊息佇列中介軟體之RabbitMQ簡介

08 如果讓你來開發一個訊息佇列中介軟體，你會怎麼設計架構？

在 CentOS7 上安裝 RabbitMQ 訊息佇列中介軟體

訊息佇列中介軟體記錄

C#分散式訊息佇列 EQueue 2.0 釋出啦

MyCat：開源分散式資料庫中介軟體

中介軟體系列ActiveMQ，Rocketmq，Rabbitmq，Kafka，Mycat讓你深入理解學習中介軟體視訊教程網盤

MySQL開源資料傳輸中介軟體架構設計實踐

RabbitMQ訊息通訊中介軟體中的那些概念

愛可生開源資料傳輸中介軟體DTLE亮相「3306π」深圳站年會

愛可生MySQL開源資料傳輸中介軟體DTLE首次技術分享

開源訊息佇列QMQ的設計與實現理念

c#開源訊息佇列中介軟體EQueue 教程

相關推薦