tangxuehua / equeue Goto Github PK

已經成功安裝 'EQueue 2.5.5.2'。
正在將 'ECommon 2.5.7' 加入 ConsoleApplication1 中。
正在解除安裝 'ECommon 2.5.7'。
已成功解除安裝 'ECommon 2.5.7'。
安裝失敗。復原中...
Install-Package : 無法安裝套件 'ECommon 2.5.7'。您正嘗試將此套件安裝到以 '.NETFramework,Version=v4.5' 為目標的專案，但該套件不包含任何與架構相容的組件參照或內容檔案。如需詳細資訊，請聯繫套件作者。
位於 行:1 字元:16

• Install-Package <<<< equeue -Version 2.5.5.2
  • CategoryInfo : NotSpecified: (:) [Install-Package], InvalidOperationException
  • FullyQualifiedErrorId : NuGetCmdletUnhandledException,NuGet.PowerShell.Commands.InstallPackageCommand

希望添加查询某Topic某队列某CurrentOffset的消息查询，方便查看堵塞的队列堆积的消息，特别是队列头上的消息，方便查找堆积原因

比如删除所有消息，删除只保留最近N天的消息，删除只保留最近N个文件的消息

1.纯in-memory模式，支持监控，不支持故障恢复，监控数据来自内存数据；由于消息都简单的放在内存，假如消费者都挂了，就会有大量消息堆积，此时就会有垃圾回收的问题，堆积的越多，就会越大的影响发送消息的tps；生产环境不建议使用这种模式；
2.in-memory+本地kv store模式（存储消息）+本地文件json形式存储消息消费进度信息，这种模式支持监控、消息堆积受内存大小限制、支持故障恢复；
3.in-memory+异步批量保存消息到db+本地WAL写日志（db存储消息和消息消费进度），这种模式支持监控、消息堆积受内存大小限制、支持故障恢复；

2,3这两种模式，本地或db中会可靠持久化消息，内存中可以缓存一定数量额消息（可配置）；当消费者要消费消息，读取消息时，从存储批量拉取消息，然后缓存到内存，然后后续消息从内存直接读取，如果内存的消息都消费完了，再继续从存储批量拉取下一批消息然后再缓存到内存中；

routingKey参数使用“623dab957e3a4333c4a4e0d9”会导致溢出。

{"Data":null,"Status":1,"ErrorMessage":"No available routing queue for topic [NoteCommandTopic]."}
这个错误要怎么处理呢？

equeueadmin中可否增加统计功能？ 比如 某个Topic，某个Group 按日统计 每小时 消费了多少条消息？

1）支持broker集群；
2）支持broker master-slave切换；

意外停电后，consume-offsets.json文件的内容可能会乱码。需要解决这个问题，可以考虑同时backup文件进行备份保护。

目前，对于消费失败（遇到异常）的消息，我们先记录错误日志，然后将该消息放入本地内存的重试队列；放入重试队列后，会定期对该消息进行重试，重试队列中的消息会定时被取出一个来重试。通过这样的设计，可以确保消费有异常的消息不会被认为消费已成功，也就是说不会从ProcessQueue中移除；但不影响该消息的后续消息的消费，该消息的后续消息仍然能够被消费，但是ProcessQueue的消费位置，即滑动门不会向前移动了；因为只要该消息一直消费遇到异常，那就意味着该消息所对应的queueOffset不能被认为已消费；而我们发送到broker的是当前最小的已被成功消费的queueOffset，所以broker上记录的当前queue的消费位置（消费进度）不会往前移动，直到当前失败的消息消费成功为止。所以，如果我们重启了消费者服务器，那下一次开始消费的消费位置还是从当前失败的位置开始，即便当前失败的消息的后续消息之前已经被消费过了；所以应用需要对每个消息的消费都要支持幂等；未来，我们会在broker上支持重试队列，然后我们可以将消费失败的消息发回到broker上的重试队列，发回到broker上的重试队列成功后，就可以让当前queue的消费位置往前移动了。

增加一个配置项。

for example:
if (!_isMemoryChunk && _chunkConfig.EnableCache)
{
var chunkSize = (ulong)GetChunkSize(_chunkHeader);
if (ChunkUtil.IsMemoryEnoughToCacheChunk(chunkSize, (uint)_chunkConfig.ChunkCacheMaxPercent))
{
try
{
_memoryChunk = Chunk.CreateNew(_filename, chunkNumber, _chunkManager, _chunkConfig, true);
}
catch (OutOfMemoryException)
{
_cacheItems = new CacheItem[_chunkConfig.ChunkLocalCacheSize];
}
catch (Exception ex)
{
_logger.Error(string.Format("Failed to cache new chunk {0}", this), ex);
_cacheItems = new CacheItem[_chunkConfig.ChunkLocalCacheSize];
}
}
else
{
_cacheItems = new CacheItem[_chunkConfig.ChunkLocalCacheSize];
}
}

如题

目前，当消费者与Broker一断开，就会立即将其从消费者列表中移除，从而导致新一轮的负载均衡时，同一个队列会被分配到另一个消费者；但有时消费者与Broker之间只是闪断，很快又会自动连上的。所以不应该因为这种闪断而改变queue对应的消费者。

所以，为了让queue和consumer之间的对应关系可以更稳定不变（除非真的增加或减少consumer服务器），去掉这个逻辑，然后只通过心跳来判断消费者是否已断开。心跳机制因为有时间方面的考虑，并不是一断开就立即认为该消费者已挂掉；而是通过判断当多少时间都没有收到consumer的心跳，才认为该consumer已断开。这个时间我们开发者可以自己配置。

IP@PORT的格式

Can you please update more on documents?

Not sure what is perfect use of this queue, we have really awsome AWS/Azure infructure for this. Any specific reason you have create this queue?

希望新版本的EQueue的集群功能在简单的使用场景下，对用户更加友好（即不需要理解NameServer等概念及部署细节，比如可以增加配置，关闭集群功能）

来个wiki呗。

您好，我Clone了一个Equeue，但是编译不过，引用需要一个ECommon的包，我看了一下，Nuget上也没有，请问能传到Nuget上吗？或者传到GIT上

考虑用fastjson

消息ID例子:"0a0acbb5881300007090330000000000"
报错原因

IPAddress初始化时接受参数如果是负数,报错信息为Specified argument was out of the range of valid values. Parameter name: newAddress
解决办法
var ip = BitConverter.ToInt32(ipBytes, 0);
改为
var ip = BitConverter.ToUInt32(ipBytes, 0);

public static MessageIdInfo ParseMessageId(string messageId)
{
var messageIdBytes = ObjectId.ParseHexString(messageId);
var ipBytes = new byte[4];
var portBytes = new byte[4];
var messagePositionBytes = new byte[8];

        Buffer.BlockCopy(messageIdBytes, 0, ipBytes, 0, 4);
        Buffer.BlockCopy(messageIdBytes, 4, portBytes, 0, 4);
        Buffer.BlockCopy(messageIdBytes, 8, messagePositionBytes, 0, 8);

        var ip = BitConverter.ToInt32(ipBytes, 0);
        var port = BitConverter.ToInt32(portBytes, 0);
        var messagePosition = BitConverter.ToInt64(messagePositionBytes, 0);

        return new MessageIdInfo
        {
            IP = new IPAddress(ip),
            Port = port,
            MessagePosition = messagePosition
        };
    }

应该还是要开启

怎么指定再思考下。