c#开源消息队列中间件EQueue 教程-低调大师

c#开源消息队列中间件EQueue 教程

2017-07-31 636

一、简介

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启动成功：

三、在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;
        }

EQueue兼容Linux/Mono，下面是CentOS 6.4/Mono 3.2.3 环境下的运行结果：

本文来自云栖社区合作伙伴“doNET跨平台”，了解相关信息可以关注“opendotnet”微信公众号

微信关注我们

原文链接：https://yq.aliyun.com/articles/176268

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

SDN、SDS催生数据中心大聚合

大型数据中心的存储流量和存储容量需求正在大幅增长。为了实现数据中心聚合，整合软件定义网络和软件定义存储是实现自动化存储分配和高效存储流量管理的重要条件。软件定义存储(SDS)支持基于软件的虚拟化、分配和数据存储管理，完全独立于底层硬件之外。SDS提供了可用于管理存储去重、复制、快照、备份及支持其他特殊存储需求的软件。计算虚拟化的增长、服务器性能的提升和廉价存储手段的流行给IT带来了大量高效部署现代数据中心存储容量的新手段。通过存储虚拟化，SDS使IT能够根据需要灵活分配存储。管理员可以在不中断应用程序访问的前提下升级物理存储硬件。此外，SDS还通过集中式管理网站去管理共享存储池，从而简化了存储的分配和管理工作。影响数据中心聚合的SDS 计算能力大幅增长(多核技术)，加上现代应用程序变得越来越复杂(包含视频、物联网[IoT]和大数据)，促使PB级大存储容量现在已经变得很常见。闪存、超聚合系统和虚拟存储区网络及Hadoop等协议给数据中心网络带来了更多的要求。存储流量现在要求东西向传输(也要南北向传输)，而且通常要求支持低延迟数据访问。数据中心网络必须支持弹性容量才能应对各种需求，...

2017-08-01

660

企业不应再局限于技术本身，而应聚焦于竞争本质的变化。然而如何实现这一模式的转变，首先得做到正确定义“智能互联”，其次是研究产品对整个行业的冲击，最后随着一个产品产生将边界模糊化那就需要产品系统进化到包含子系统的产品体系在智能互联产品时代，企业该如何创造和捕捉价值?产品产生的(高度敏感的)海量数据应该如何利用和管理?如何改进与传统业务伙伴，例如渠道商之间的关系?随着行业边界的极大拓展，公司在其中应该扮演什么样的角色?企业将面临上述一系列新的战略抉择。随着智能互联产品数量不断增多，为了阐释随之而来的新机遇，“物联网”一词应运而生。但它的诞生无助于我们理解这一现象及其影响。无论是涉及物或人，互联始终是一种传递信息的机制。智能互联产品的独特之处不在于互联，而在于“物”，正是产品的新能力其产生的数据将开创一个新的竞争时代。因此，企业不应再局限于技术本身，而应聚焦于竞争本质的变化。定义“智能互联” 智能互联产品包含3个核心元素：物理部件、智能部件和联接部件。智能部件能加强物理部件的功能和价值，而联接部件进一步强化智能部件的功能和价值，并让部分价值和功能脱离物理产品本身存在。这就使得价值提升形...

2017-08-01

615

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。