RawMessage 是从graylog input 组件获取到的还没解析处理的消息，是有SimpleChannelInboundHandler 处理的原始消息
RawMessageEvent 是Disruptor 处理的事件包装
MessageEvent 是经过input 之后到ProcessBuffer 中的消息事件（Disruptor 包装的）
Message 是经过input 以及ProcessBufferProcessor 之后生成的消息
参考图
MessageEvent 是包含了RawMessage 以及message经过Disruptor 的事件
RawMessage参考代码
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
final byte[] bytes = new byte[msg.readableBytes()];
msg.readBytes(bytes);
final RawMessage raw = new RawMessage(bytes, (InetSocketAddress) ctx.channel().remoteAddress());
input.processRawMessage(raw); // 会通过MessageInput 进行处理，会放到inputbuffer 中
}
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
RawMessageEvent 参考代码
@Inject
public InputBufferImpl(MetricRegistry metricRegistry,
BaseConfiguration configuration,
Provider<DirectMessageHandler> directMessageHandlerProvider,
Provider<RawMessageEncoderHandler> rawMessageEncoderHandlerProvider,
Provider<JournallingMessageHandler> spoolingMessageHandlerProvider) {
// 基于Disruptor的事件包装，是由 RawMessage 产生的，具体可以通过
final Disruptor<RawMessageEvent> disruptor = new Disruptor<>(
RawMessageEvent.FACTORY,
configuration.getInputBufferRingSize(),
threadFactory(metricRegistry),
ProducerType.MULTI,
configuration.getInputBufferWaitStrategy());
disruptor.setDefaultExceptionHandler(new LoggingExceptionHandler(LOG));
final int numberOfHandlers = configuration.getInputbufferProcessors();
if (configuration.isMessageJournalEnabled()) {
("Message journal is enabled.");
final RawMessageEncoderHandler[] handlers = new RawMessageEncoderHandler[numberOfHandlers];
for (int i = 0; i < numberOfHandlers; i++) {
handlers[i] = rawMessageEncoderHandlerProvider.get();
}
disruptor.handleEventsWithWorkerPool(handlers).then(spoolingMessageHandlerProvider.get());
} else {
("Message journal is disabled.");
final DirectMessageHandler[] handlers = new DirectMessageHandler[numberOfHandlers];
for (int i = 0; i < numberOfHandlers; i++) {
handlers[i] = directMessageHandlerProvider.get();
}
disruptor.handleEventsWithWorkerPool(handlers);
}
ringBuffer = disruptor.start();
incomingMessages = metricRegistry.meter(name(InputBufferImpl.class, "incomingMessages"));
safelyRegister(metricRegistry, GlobalMetricNames.INPUT_BUFFER_USAGE, new Gauge<Long>() {
@Override
public Long getValue() {
return InputBufferImpl.this.getUsage();
}
});
safelyRegister(metricRegistry, GlobalMetricNames.INPUT_BUFFER_SIZE, constantGauge(ringBuffer.getBufferSize()));
("Initialized {} with ring size <{}> and wait strategy <{}>, running {} parallel message handlers.",
this.getClass().getSimpleName(),
configuration.getInputBufferRingSize(),
configuration.getInputBufferWaitStrategy().getClass().getSimpleName(),
numberOfHandlers);
}
public InputBufferImpl(MetricRegistry metricRegistry,
MessageEvent 参考处理
ProcessBuffer.java 通过inputbuffer 放到 ProcessBuffer 中
ProcessBuffer.java 通过inputbuffer 放到 ProcessBuffer 中
public void insertBlocking(@Nonnull RawMessage rawMessage) {
final long sequence = ringBuffer.next();
final MessageEvent event = ringBuffer.get(sequence);
event.setRaw(rawMessage);
ringBuffer.publish(sequence);
afterInsert(1);
}
Message 提供的能力
后续MessageProcessor 处理的就是此类，同时后续的output 也是使用的此消息进行存储处理
