百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 编程网 > 正文

Disruptor框架源码阅读-如何不重复消费

yuyutoo 2025-03-30 22:53 3 浏览 0 评论

RingBuffer 如何保证数据不丢失

由于ringbuffer是一个环形的队列,那么生产者和消费者在遍历这个队列的时候,如何制衡呢?

1、生产快,消费慢,数据丢失?

生产者速度过快,导致一个对象还没消费完,就循环生产了一个新的对象要加入ringbuffer,导致消费不完整,造成数据丢失?

我们注意到,在我们获取生产者下一个位置的时候,是通过ringbuffer的next方法,而这个next方式是调用了sequencer的next方法

这个对象,在我们创建disruptor对象的时候,创建的

所以这个ringbuffer就是disruptor中的sequencer对象,那么在进行获取next的时候,这里是如何获取下一个的呢?是否会对这个生产获取下一个序列进行相应的等待策略,避免产生相应的干扰!!!

这个各位看官还需多看看里面的代码以及封装(特别是封装,真是九转十八弯),多熟悉,我这绕着绕着很容易就绕晕了,刚开始也是云里雾里。

EventProcessor接口概览

EventProcessor顾名思义,就是事件处理器(handle和process都可以翻译为“处理”,但是process侧重于机器的处理,而handle侧重于有人工的处理,所以使用handle表示用户逻辑的处理,使用process表示机器的处理),这个接口有两个实现类,分别是WorkProcessor和BatchEventProcessor,它们对应的逻辑处理消费者分别是EventHandler和WorkHandler。下面是EventProcessor的UML类图及EventHandler和EventProcessor的接口定义。



/**
* Callback interface to be implemented for processing events as they become available in the {@link RingBuffer}
*
* @param  event implementation storing the data for sharing during exchange or parallel coordination of an event.
* @see BatchEventProcessor#setExceptionHandler(ExceptionHandler) if you want to handle exceptions propagated out of the handler.
* 处理事件的回调接口
*/
public interface EventHandler
{
 /**
 * Called when a publisher has published an event to the {@link RingBuffer}
 *
 * @param event published to the {@link RingBuffer}
 * @param sequence of the event being processed
 * @param endOfBatch flag to indicate if this is the last event in a batch from the {@link RingBuffer}
 * @throws Exception if the EventHandler would like the exception handled further up the chain.
 */
 void onEvent(T event, long sequence, boolean endOfBatch) throws Exception;
}
/**
* EventProcessors waitFor events to become available for consumption from the {@link RingBuffer}
* 

* An EventProcessor will generally be associated with a Thread for execution. * 事件执行器,等待RingBuffer有可用消费事件。一个事件处理器关联一个执行线程 */ public interface EventProcessor extends Runnable { /** * Get a reference to the {@link Sequence} being used by this {@link EventProcessor}. * * @return reference to the {@link Sequence} for this {@link EventProcessor} */ Sequence getSequence(); /** * Signal that this EventProcessor should stop when it has finished consuming at the next clean break. * It will call {@link SequenceBarrier#alert()} to notify the thread to check status. */ void halt(); boolean isRunning(); }

EventProcessor接口继承了Runnable接口,主要有两种实现:单线程批量处理BatchEventProcessor和多线程处理WorkProcessor

在使用Disruptor帮助类构建消费者时,使用handleEventsWith方法传入多个EventHandler,内部使用多个BatchEventProcessor关联多个线程执行。这种情况类似JMS中的发布订阅模式,同一事件会被多个消费者并行消费。适用于同一事件触发多种操作。

而使用Disruptor的
handleEventsWithWorkerPool传入多个WorkHandler时,内部使用多个WorkProcessor关联多个线程执行。这种情况类似JMS的
点对点模式,同一事件会被一组消费者其中之一消费。适用于提升消费者并行处理能力。

消费技术实现

我们先回顾下Disruptor消费者的两个特点:消费者依赖图(即下文所谓的“消费链”)和事件多播。

假设现在有A,B,C,D四个消费者,它们都能组成什么样的形式呢?从众多的排列组合中,我挑了4组比较有代表性的消费链形式。



image.png

  • 第1组中,消费者A消费按成后,B、C、D可同时消费;
  • 第2组中,消费者A、B、C、D顺序消费;
  • 第3组中,消费者A、B顺序消费后,C、D同时消费;
  • 第4组中,消费者A在消费完成后,B和C可以同时消费,但是必须在都消费完成后,D才能消费。

标号为1、3、4的消费链都使用了事件多播,可见事件多播属于消费链的一种组合形式。注意,在上面4种组合中,每个组合的每一水平行,都属于一个消费者组。

这些还只是较为简单的消费链组成,实际中消费链可能会更复杂。

那么在Disruptor内部是怎么实现消费链的呢?

我们可以先思考下。如果想把独立的消费者组成消费链,那么后方的消费者(组)必然要知道在它前方的消费者(组)的处理情况,否则就做不到顺序消费。同时,消费者也要了解生产者的位置,来判断是否有可用事件。之前我们分析生产者代码的时候,已经讲过,生产者为了不覆盖没有消费完全的事件,必须知道最慢消费者的处理情况

做到了这些才会有能力去控制消费者组成消费链。下面让我们具体看Disruptor中的实现。

单生产者,多消费者模式。多消费者对于消息不重复消费。

package liuqiang.complex.multi;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import liuqiang.complex.common.*;
import java.util.concurrent.Executors;
public class Main3 {
 //单生产者,多消费者模式。多消费者对于消息不重复消费。例如:1线程消费了消息0,则2线程只能从0后面的消息消费,不能对消息0进行消费。
 public static void main(String[] args) throws Exception {
 EventFactory factory = new OrderFactory();
 int ringBufferSize = 1024 * 1024;
 Disruptor disruptor =
 new Disruptor(factory, ringBufferSize, Executors.defaultThreadFactory(), ProducerType.SINGLE, new YieldingWaitStrategy());
 /*
 * 该方法传入的消费者需要实现WorkHandler接口,方法的内部实现是:先创建WorkPool,然后封装WorkPool为EventHandlerPool返回。
 * 消费者1、2对于消息的消费有时有竞争,保证同一消息只能有一个消费者消费
 */
 disruptor.handleEventsWithWorkerPool(new OrderHandler1("1"), new OrderHandler1("2"));
 disruptor.start();
 RingBuffer ringBuffer = disruptor.getRingBuffer();
 Producer producer = new Producer(ringBuffer);
 //单生产者,生产3条数据
 for (int l = 0; l < 3; l++) {
 producer.onData(l + "");
 }
 //为了保证消费者线程已经启动,留足足够的时间。具体原因详见另一篇博客:disruptor的shutdown失效问题
 Thread.sleep(1000);
 disruptor.shutdown();
 }
}

调用
handleEventsWithWorkerPool形成WorkerPool,并进一步封装成EventHandlerGroup。对于同一条消息,两消费者不重复消费。

可能输出结果如下:

OrderHandler1 1,消费信息:0

OrderHandler1 2,消费信息:1

OrderHandler1 1,消费信息:2

消费者可用序列屏障-SequenceBarrier

我们重点看一下SequenceBarrier,可直译为“序列屏障”。SequenceBarrier的主要作用是协调获取消费者可处理到的最大序号,内部持有着生产者和其依赖的消费者序列。它的接口定义如下。

public interface SequenceBarrier
{
 /**
 * Wait for the given sequence to be available for consumption.
* 等待指定序列可用 * @param sequence to wait for * @return the sequence up to which is available * @throws AlertException if a status change has occurred for the Disruptor * @throws InterruptedException if the thread needs awaking on a condition variable. * @throws TimeoutException * */ long waitFor(long sequence) throws AlertException, InterruptedException, TimeoutException; /** * Get the current cursor value that can be read.
* 获取当前可读游标值 * * @return value of the cursor for entries that have been published. * */ long getCursor(); /** * The current alert status for the barrier.
* 当前的alert状态 * * @return true if in alert otherwise false. */ boolean isAlerted(); /** * Alert the {@link EventProcessor}s of a status change and stay in this status until cleared.
* * 通知消费者状态变化。当调用EventProcessor#halt()将调用此方法。 */ void alert(); /** * Clear the current alert status.
* 清楚alert状态 */ void clearAlert(); /** * Check if an alert has been raised and throw an {@link AlertException} if it has. * 检查是否发生alert,发生将抛出异常 * @throws AlertException if alert has been raised. */ void checkAlert() throws AlertException; } SequenceBarrier实例引用被EventProcessor持有,用于等待并获取可用的消费事件,主要体现在waitFor这个方法。

要实现这个功能,需要3点条件:

  1. 知道生产者的位置。
  2. 因为Disruptor支持消费者链,在不同的消费者组之间,要保证后边的消 费者组只有在前消费者组中的消费者都处理完毕后,才能进行处理。
  3. 暂时没有事件可消费,在等待可用消费时,还需要使用某种等待策略进行等待。

看下SequenceBarrier实现类ProcessingSequenceBarrier的代码是如何实现waitFor方法。

final class ProcessingSequenceBarrier implements SequenceBarrier
{
 private final WaitStrategy waitStrategy; // 等待可用消费时,指定的等待策略
 private final Sequence dependentSequence; // 依赖的上组消费者的序号,如果当前为第一组则为cursorSequence(即生产者发布游标序列),否则使用FixedSequenceGroup封装上组消费者序列
 private volatile boolean alerted = false; // 当触发halt时,将标记alerted为true
 private final Sequence cursorSequence; // AbstractSequencer中的cursor引用,记录当前发布者发布的最新位置
 private final Sequencer sequencer; // MultiProducerSequencer 或 SingleProducerSequencer
 public ProcessingSequenceBarrier(
 final Sequencer sequencer,
 final WaitStrategy waitStrategy,
 final Sequence cursorSequence,
 final Sequence[] dependentSequences)
 {
 this.sequencer = sequencer;
 this.waitStrategy = waitStrategy;
 this.cursorSequence = cursorSequence;
 if (0 == dependentSequences.length) // 依赖的上一组序列长度,第一次是0
 {
 dependentSequence = cursorSequence;
 }
 else // 将上一组序列数组复制成新数组保存,引用不变
 {
 dependentSequence = new FixedSequenceGroup(dependentSequences);
 }
 }
 @Override
 public long waitFor(final long sequence)
 throws AlertException, InterruptedException, TimeoutException
 {
 // 检查是否停止服务
 checkAlert();
 // 获取最大可用序号 sequence为给定序号,一般为当前序号+1,cursorSequence记录生产者最新位置,
 long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
 if (availableSequence < sequence)
 {
 return availableSequence;
 }
 // 返回已发布最高的序列值,将对每个序号进行校验
 return sequencer.getHighestPublishedSequence(sequence, availableSequence);
 }
 // ... 
}

相关推荐

YAML配置文件简介及使用(yaml 配置)

简介YAML是"YAMLAin'taMarkupLanguage"(YAML不是一种标记语言)的缩写。相比JSON格式的方便。...

教你如何解决最常见的58种网络故障排除方法

1.故障现象:网络适配器(网卡)设置与计算机资源有冲突。分析、排除:通过调整网卡资源中的IRQ和I/O值来避开与计算机其它资源的冲突。有些情况还需要通过设置主板的跳线来调整与其它资源的冲突。2.故障现...

一分钟带你了解服务器网卡(服务器网卡怎么用)

今天小编和大家聊一下服务器的网卡。什么是网卡?简单说网卡就是计算机与局域网互连的设备。计算机主要通过网卡接入网络。网卡又称为网络适配器或网络接口卡NIC(NetworkinterfaceCard)...

linux文件之ssh配置文件的含义与作用

ssh远程登录命令是操作系统(包括linux和window系统)下常用的操作命令,可以帮助用户,远程登录服务器系统,查看,操作系统相关信息。linux系统对于ssh命令有专门保存其相关配置的目录和文件...

Cilium 官方文档翻译 - IPAM(二)Kubernetes Host模式

KubernetesHostScopeciliumIPAM的kuberneteshost-scope模式通过选项ipam:kubernetes开启,将集群IP地址分配委托给每个独立的节点,并...

域名劫持跳转,域名劫持跳转的解决办法只需5步

简单来说,域名劫持就是把原本准备访问某网站的用户,在不知不觉中,劫持到仿冒的网站上,例如用户准备访问某家知名品牌的网上商店,黑客就可以通过域名劫持的手段,把其带到假的网上商店,同时收集用户的ID信息和...

Linux基本命令(linux基本命令总结)

...

Linux 磁盘和文件系统管理(linux磁盘管理fdisk)

1检测并确认新硬盘...

windows host文件怎么恢复?局域网访问全靠这些!

windowshost文件怎么恢复?windowshost文件是常用网址域名及其相应IP地址建立一个关联文件,通过这个host文件配置域名和IP的映射关系,以提高域名解析的速度,方便局域网用户使用...

Nginx配置文件详解与优化建议(nginx 配置详解)

1、概述今天来详解一下Nginx的配置文件,以及给出一些配置建议,希望能对大家有所帮助。...

Mac电脑hosts文件锁定,如何修改hosts文件权限

有时候我们需要修改hosts文件,但是网上很多教程都行不通,使用sudo命令也不行。其实有一个很简单的方法。打开终端命令行,使用如下命令即可:sudochflags-hvnoschg/etc/...

windows电脑如何修改hosts文件?(windows 修改hosts文件)

先来简单说下电脑host的作用hosts文件的作用:hosts文件是一个用于储存计算机网络中各节点信息的计算机文件;作用是将一些常用的网址域名与其对应的IP地址建立一个关联“数据库”,当用户在浏览器中...

Vigilante恶意软件行为怪异:修改Hosts文件以阻止受害者访问盗版网站

Sophos刚刚报道了一款名叫Vigilante的恶意软件,但其行为却让许多受害者感到不解。与其它专注于偷密码、搞破坏、或勒索赎金的恶意软件不同,Vigilante会通过修改Hosts文件...

hosts文件无法修改几种现象和解决方法

第一种、hosts文件修改完不是直接保存而是弹出另存为窗口解决:1、右击hosts文件——属性——把“只读”前面勾去掉。第二种、打开hosts文件时提示“你没有权限打开该文件,请向文件的所有者或管理员...

hosts文件位置在哪里,教你hosts文件位置在哪里

Hosts是一个没有扩展名的系统文件,其基本作用就是将一些常用的网址域名与其对应的IP地址建立一个关联"数据库",当用户在浏览器中输入一个需要登录的网址时,系统会首先自动从Hosts文件中寻找对应的I...

取消回复欢迎 发表评论: