标签: message-queue

我正试着绕过AMQP.它看起来很适合应用程序之间的机器间(集群,LAN,WAN)通信,但我不确定它是否适合(在架构和当前实现术语中)用作一台机器中的软件总线.

是否值得提取当前的高性能消息传递框架以用AMQP替换它,或者是否通过模糊本地和非本地通信之间的区别而陷入与RPC相同的陷阱？

我也对使用WAN技术进行机内通信的性能影响保持警惕,尽管这可能比架构更像是一个实现问题.

战争故事将不胜感激.

我目前正在设计一个我最终想要迁移到Windows Azure的应用程序.但是,在短期内,它将在我自己托管的服务器上运行.

该应用程序涉及许多单独的Web应用程序 - 其中一些本质上是接收数据的WCF服务,一些是用户管理数据的站点.此外,还需要在后台运行工作服务,以各种方式处理数据.

我非常渴望为此使用解耦架构.理想情况下,我希望组件(即Web应用程序和工作服务)尽可能少地了解彼此.似乎使用消息队列将是最好的解决方案 - Web应用程序可以将带有工作单元的消息排入队列,工作服务可以将它们挑选出来并根据需要进行处理.

但是,我想制定一套很好的技术来做到这一点,记住我最终会转向Azure,并希望尽量减少迁移到云时我需要做的工作量.Azure内置了一个Queue组件,看起来非常适合我的需求.我想做的是自己创造一些尽可能模仿的东西.

看起来有几个选项(我在Windows上使用.NET,带有SQL Server 2005后端) - 到目前为止我发现的是:

• MSMQ
• SQL Server服务代理
• 使用数据库表和一些存储过程滚动我自己

我想知道是否有人对此有任何建议 - 或者是否有人做了类似的事情,并就要做/要避免的事情提出建议.我意识到每种情况都不同,但在这种情况下,我认为我的排队要求非常通用,所以我很想听到别人对最佳方法的看法.

提前致谢,

约翰

在54:53分钟的谈话中,Rich Hickey正在谈论使用队列作为解耦依赖程序部分的手段.你能给我一个例子,说明如何将下面的Java伪代码分离出来,以便改进它的设计和/或灵活性:

// Warning: Java-pseudo-code ahead
class Job {
    public void doRun(A a) {
        saveObjectToDatabase(a);

        B b = computeB(a);
        saveObjectToDatabase(b);

        C c = computeC(b);
        logToFile(c);
    }
}

saveObjectToDatabase并且saveObjectToDatabase可以看作具有副作用的方法,而computeB's和computeC's输出仅依赖于a.

我知道这个问题相当模糊/广泛.我想了解如何利用排队机制,而不会使我的程序大规模复杂化,并仍然确保它以正确的顺序做正确的事情.任何指向正确方向的人都会受到赞赏.

我正在考虑系统将通知多个消费者有关一群物体发生的事件.每个订阅者应该能够订阅发生在零个或多个对象上的事件,多个订阅者应该能够接收有关发生在单个对象上的事件的信息.

我认为在这种情况下,某些消息排队系统是合适的,但我不知道如何处理我将拥有数百万个对象的事实 - 对每个对象使用单独的主题听起来不太好[或者是它正好？].

你能否建议我应该采取的方法,甚至可能是一些合理的开源消息排队系统？

更多细节:

• 会有成千上万的订阅者[意思不是很多]
• 订阅者将订阅每个数十或数百个对象,
• 将有约5-20万个物体,
• 事件本身不必携带任何信息.只是该对象被更改的信息就足够了,
• 绝大多数对象永远不会订阅,
• 事件以每秒几百个的最大速率发生,
• 理想情况下,服务器应该在linux下运行,能够通过http long-poll与生态系统的其余部分集成[使用节点js？码头下的延续？].

在此先感谢您的反馈,并抱歉有些模糊的问题!

每秒只有20条消息!这就是我得到的!这是从队列中查看50条消息并使用ReceiveById并行接收它们的代码.队列中的消息总数是500.我也测试了其他数字.但是上限是每秒20条消息!我在某处完全不受影响吗？

编辑1:

1 - 我需要队列可以恢复.但有趣的是,即使我将可恢复选项设置为false; 仍然是最高限制是20消息/秒.

2 - 我被迫在这里使用MSMQ,因为涉及一些遗留应用程序.但是如果这个代码是正确的并且这个前20个限制确实存在,我可以说服该组切换.因此,任何推荐(基于实际经验)的替换MSMQ都是非常受欢迎的(请注意,如果出现任何类型的失败,我们需要保留我们的消息).

3 - 我已经将ThreadPool中的线程数设置为一个高数字,以防它有所帮助但实际上在这段代码中它将导致创建100 - 200个线程.我已经测试了从50到10000的不同数字,没有差异.

4 - 在每个任务中创建一个新的MessageQueue,因为ReceiveById不是线程安全的.

5 - 正如人们在代码中看到的那样,消息大小非常低; 它只是一个字符串加一个int.

编辑2:[ 非常奇怪的新结果 ]

我已经玩了这段代码的每一点,发现了这个:如果我注释掉行singleLocal.UseJournalQueue = false; 在我的任务中,我每秒最多可以读取1200条消息.不是很令人印象深刻,但在我的情 奇怪的部分是UseJournalQueue的默认值为false; 为什么再将它设置为假应该在性能上有所不同？

static partial class Program
{
    static void Main(string[] args)
    {
        ThreadPool.SetMaxThreads(15000, 30000);
        ThreadPool.SetMinThreads(10000, 20000);

        var qName = @".\private$\deep_den";

        if (!MessageQueue.Exists(qName))
        {
            var q = MessageQueue.Create(qName);
        }

        var single = new MessageQueue(qName);
        single.UseJournalQueue = false;
        single.DefaultPropertiesToSend.AttachSenderId = false;
        single.DefaultPropertiesToSend.Recoverable = true;
        single.Formatter = new XmlMessageFormatter(new[] { typeof(Data) });

        var count …

我刚刚起步并运行Kafka 0.8 beta 1.我有一个非常简单的示例启动和运行,问题是,我只能让一个消息消费者工作,而不是几个.也就是说,runSingleWorker()方法可以工作.run()方法不起作用:

import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.consumer.ConsumerConfig;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.Map;
import java.util.List;
import java.util.HashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import com.truecar.inventory.worker.core.application.config.AppConfig;

public class ConsumerThreadPool {

    private final ConsumerConnector consumer;
    private final String topic;

    private ExecutorService executor;
    private static ApplicationContext context = new AnnotationConfigApplicationContext(AppConfig.class);

    public ConsumerThreadPool(String topic) {
        consumer = kafka.consumer.Consumer.createJavaConsumerConnector((ConsumerConfig)context.getBean("consumerConfig"));
        this.topic = topic;
    }

    public void shutdown() {
        if (consumer != null) consumer.shutdown();
        if (executor != null) executor.shutdown();
    }

    public void run(Integer numThreads) …

我正在开发一个Android应用程序,它从Internet获取图像并在用户界面中显示.我正在使用RecyclerView来显示图像.我打算使用单独的线程下载图像.并通过处理程序更新RecyclerView.我不知道这个概念是否正确,(我知道AsyncTask,但出于学习目的,我正在尝试实现Handler.) 所以我的编码与下面相同

private void loadNewsThumbnailImage(ArrayList<DataItem> dataList) {
    for (DataItem item : DataList) { //DataItem is the model class
        loadThumbnailFromInternet(item);
        }
}



private void loadThumbnailFromInternet(final DataItem dataItem) {

        Thread imageDowloaderThread = new Thread(new Runnable() {
            @Override
            public void run() {
                Bitmap bitmap = null;
                try {
                    bitmap = getDataItemBitmap(dataItem.getmImageUrl());
                    dataItem.setmThumbnail(bitmap);
                    new Handler().post(new Runnable() { // Tried new Handler(Looper.myLopper()) also
                        @Override
                        public void run() {
                            mAdapter.notifyDataSetChanged();
                        }
                    });
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
        imageDowloaderThread.start();

    }

我已执行此代码,但我收到错误,应用程序终止,我不知道为什么会发生这种情况.请任何人帮我解决一下.并解释当前代码的问题.

(请不要建议使用AsyncTask(我已经尝试过,它工作正常)) …

在试图深入了解 Kafka 分发模型时，StackOverflow 的一句话让我兴奋不已，我无法得到确认或否认。

因此，订阅者组越多，性能就越低，因为 kafka 需要将消息复制到所有这些组并保证总​​顺序。

据我从 Kafka 文档中了解到，多个消费者组的行为与单个消费者类似。代理内不会进行复制，因为每个消费者对于某个分区都有自己的偏移量。那么，组的数量不应产生任何显着的开销，所有数据都位于一个位置，只是偏移量不同。那是对的吗？

如果这是正确的，那么实际上无法在不影响吞吐量的情况下引入多个不相交的消费者，因为所有消费者总是查询所有分区，并且引入了某种复制。请注意，这与消费者线程的数量无关，线程只会提高消费者性能，据我所知，它们不会干扰代理操作。

我目前面临一个问题，我可以肯定有一个正式名称，但是我不知道该在网上搜索什么。我希望如果我描述问题和解决方案时想到的，有人能够告诉我设计模式的名称（如果有一个与我要描述的内容相匹配的设计模式）。

基本上，我想拥有一个工作队列：我有多个创建工作的客户端（发布者），以及许多处理这些工作的工人（消费者）。现在，我想将发布者创建的作业分发给各个使用者，这基本上可以使用几乎任何消息队列并在整个队列中实现负载均衡，例如使用RabbitMQ甚至MQTT 5。

但是，现在事情变得复杂了……每个工作都指向一个外部实体，例如一个用户。我想要的是按顺序处理单个用户的作业，但并行处理多个用户的作业。我不要求用户X的作业始终去工作者Y，因为无论如何它们都应按顺序处理。

现在，我可以使用RabbitMQ及其一致的哈希交换来解决此问题，但是当新的工作人员进入集群时，我就会进行数据竞赛，因为RabbitMQ不支持重新定位已经在队列中的作业。

MQTT 5也不支持：在这里，这个想法被称为“粘性共享订阅”，但这不是官方的。它可能是MQTT 6的一部分，也可能不是。谁知道。

我也看过NSQ，NATS和其他一些经纪人。他们中的大多数甚至都不支持这种非常特定的情况，而那些确实使用一致哈希的情况，则存在前面提到的数据竞速问题。

现在，如果代理在作业到达后不将作业分类到队列中，而是跟踪某个特定用户的作业是否已经在处理中，则问题将消失：如果这样，它将延迟所有其他作业该用户，但其他用户的所有作业仍应处理。使用RabbitMQ等人无法做到这一点。

我很确定我不是唯一拥有用例的人。例如，我可以想到用户将视频上传到视频平台，尽管上传的视频是并行处理的，但单个用户上传的所有视频都是按顺序处理的。

因此，简而言之：我所形容的名字是用一个普通名字吗？诸如分布式作业队列之类的东西？具有任务相似性的任务调度程序？还是其他？我尝试了很多术语，但没有成功。这可能意味着没有解决方案，但是如上所述，很难想象我是这个问题上唯一的人。

有什么想法我可以寻找吗？并且：是否有实现此目的的工具？有协议吗？

PS：仅使用预定义的路由密钥是不可行的，因为用户ID（我在这里只是作为一个示例）基本上是UUID，因此可以有数十亿个，因此我需要更多动态的东西。因此，一致性哈希基本上是正确的方法，但是如上所述，为了避免数据争用，分发必须逐个进行而不是预先进行。

我正在寻找最快的方法（就延迟而言）在两个进程之间进行通信，告知事件已发生。

更准确地说，我在共享内存中有 numpy 数组，其中一个进程（生产者）将更新写入数组，另一个进程（消费者）读取它们。

需要使用多处理，因为我们需要克服 GIL。生产者是 CPU/IO 密集型进程，它监听数据流并进行一些数据处理。

Consumer 是一个非常轻量且大部分空闲的进程，但是当 Producer 更新数组时我们需要尽快唤醒它。

还有一件事。以最小的延迟触发消费者比传输所有消息更重要。（例如，如果生产者连续发送三个消息而没有延迟，而消费者仅收到第一个消息并丢失了后面的两个消息 - 没关系。）

为此，我尝试了多处理原语 Pipe、Queue、Event，看起来它们在延迟方面几乎相同。管道是最稳定的。

多处理.管道

import multiprocessing as mp
import numpy as np
import random
import time

ITER_COUNT = 1000


def get_mcs_diff(ts):
    return round((time.time() - ts) * 1e6, 0)


def main(v, input_pipe):
    for _ in range(ITER_COUNT):
        v.value = time.time()
        input_pipe.send(None)
        time.sleep((0.1 + random.random()) / 100)


if __name__ == "__main__":
    v = mp.Value('d', time.time())
    (ip, op) = mp.Pipe()
    p = mp.Process(target=main, args=(v, ip,))
    measurements = []

    p.start()
    i …