[Storm] 内部消息缓存

这篇文件翻译自 http://www.michael-noll.com/blog/2013/06/21/understanding-storm-internal-message-buffers/

当进行Storm调优时，理解Storm内部消息队列的配置十分有帮助。这篇文件将说明在Storm 0.8/0.9版本中一个Worker内部的消息通信。

Storm Worker进程内部消息传输

这里所说的“内部消息”是指单台节点上的一个Worker进程内部的消息。这种通信依赖于Storm内部各种 LMAX Disrupter（高性能多线程消息库） 支持的消息队列。

单个Worker进程内部多线程通信不同于多个worker进程之间的通信。后者通常跨网络或者机器，Storm采用 Zero MQ/Netty进行支持。

• Intra-worker communication in Storm (inter-thread on the same Storm node): LMAX Disruptor
• Inter-worker communication (node-to-node across the network): ZeroMQ or Netty
• Inter-topology communication: nothing built into Storm, you must take care of this yourself with e.g. a messaging system such as Kafka/RabbitMQ, a database, etc.

事例说明

Storm内部消息队列概要。Worker进程的消息队列标识为红色，Worker内部Executer的消息队列标识为绿色。图中为了可读性，只有一个Worker和一个Executor，而现实中通常有多个Worker，一个Worker中也可能有多个Executors。

详细描述

Worker进程

为了管理进来和出去的消息，每个Worker进程有一个监听TCP端口（configured by supervisor.slots.ports）的接收线程。类似的，每个Worker有一个发送线程负责将从transfer queue中读到的消息发送到下游消费者。

• The topology.receiver.buffer.size is the maximum number of messages that are batched together at once for appending to an executor’s incoming queue by the worker receive thread (which reads the messages from the network) Setting this parameter too high may cause a lot of problems (“heartbeat thread gets starved, throughput plummets”). The default value is 8 elements, and the value must be a power of 2 (this requirement comes indirectly from LMAX Disruptor).

• // Example: configuring via Java API
  Config conf = new Config();
  conf.put(Config.TOPOLOGY_RECEIVER_BUFFER_SIZE, 16); // default is 8

• Each element of the transfer queue configured with topology.transfer.buffer.size is actually a list of tuples. The various executor send threads will batch outgoing tuples off their outgoing queues onto the transfer queue. The default value is 1024 elements.

• // Example: configuring via Java API
  conf.put(Config.TOPOLOGY_TRANSFER_BUFFER_SIZE, 32); // default is 1024

Executors

一个Worker进程控制着一个或者多个Executer线程。每个Executor有自己的 incoming queue 和 outgoing queue

每个Executor有一个负责逻辑处理spout/bolt的线程， 一个接收消息线程，和一个发送消息线程。

• The topology.executor.receive.buffer.size is the size of the incoming queue for an executor. Each element of this queue is a list of tuples. Here, tuples are appended in batch. The default value is 1024 elements, and the value must be a power of 2 (this requirement comes from LMAX Disruptor).

• // Example: configuring via Java API
  conf.put(Config.TOPOLOGY_EXECUTOR_RECEIVE_BUFFER_SIZE, 16384); // batched; default is 1024

• The topology.executor.send.buffer.size is the size of the outgoing queue for an executor. Each element of this queue will contain a single tuple. The default value is 1024 elements, and the value must be a power of 2 (this requirement comes from LMAX Disruptor).

• // Example: configuring via Java API
  conf.put(Config.TOPOLOGY_EXECUTOR_SEND_BUFFER_SIZE, 16384); // individual tuples; default is 1024