• 起因
• 需求场景
• 思路-TODO
• 解决方案-TODO

Cause

过滤水机售卖，充值4元后进行灌水，当剩余水量0.1L时，再充值2元。此时面板上剩余的水量会变成6元的水量，触发BUG

Scenes

类似于游泳池一边放水一边加水问题。

如何保证水管仍然持续出水的情况下，第二次充值时，单次通信情况下水量正确增加同时因为水管仍放水，继续正确扣减水量

• 记录刷题记录相关
• Base Structure
• Algorithm
• LeetCode
• Other

• 基本概念
• 底层算法
• TODO状态

Java面试题

• Specification
• Java
• Spring
• MQ
• Cache
• Database
• ES
• Network
• Tools
• Theory

Transaction

• JDBC事务
• Spring 声明式事务
• Spring 编程式事务
• Spring 声明式事务-事务失效

Cloud Computing

Introduction

LMAX Disruptor

是一个高性能的线程间消息传递库（即同一进程内线程间的内存队列）。它旨在为异步事件处理体系结构提供低延迟、高吞吐量的工作队列。

Disruptor 与 Java BlockingQueue类似。

Disruptor 与队列一样，Disruptor的目的是在同一进程内的多个线程之间移动数据（例如消息或事件）。 但是，Disruptor提供了一些关键功能，可以将其与队列区分开来。 它们是：

• 向使用者发送多播事件，并带有使用者依赖关系图。
• 为事件预分配内存。
• 可选无锁。

MQ Middleware

MQ中间件使用场景一般在系统和系统间的消息传递，吞吐量高，也适用于消息流数据处理的Middleware。

Diff

• Disruptor是内存数据，MQ Middleware如Kafka数据是磁盘落库
• Disruptor是同一台Instance内部的多线程消费数据，Kafka可以提供多分片的多节点批量消费

Why Disruptor

Java内部传统队列用锁保证线程安全，而锁需要通过操作系统上下文切换实现，会暂停线程等待锁释放。

频繁切换上下文会丢失之前保存的数据和指令。由于消费者和生产者之间的速度差异，队列总是接近满或者空的状态，这种状态会导致高水平的写入争用。

Core Concepts

Ring Buffer

曾是核心功能，3.0版本后只负责对通过Disruptor进行交换的数据(Event)进行存储与更新。

对于一些高级使用场景，可以完全由用户替换。

Sequence

Disruptor 使用序列作为一种方法来识别特定组件的位置。每个消费者 (Event Processor)都维护一个 Sequence，Disruptor 本身也是如此。大多数并发代码都依赖于这些 Sequence 值的移动，因此 Sequence 支持 AtomicLong 的许多当前功能。事实上，两者之间唯一真正的区别是 Sequence 包含额外的功能(Padding)来防止 Sequences 和其他值之间的错误共享（伪共享）。

Sequencer

Disruptor的真正核心，主要是用来协调生产者。

该接口的两个实现 (single producer, multi producer) 实现了所有并发算法(concurrent algorithms)，以便在生产者和消费者之间快速、正确地传递数据。

在生产者生产数据的时候，Sequencer会产生一个可用的序号（Sequence），然后生产者就就知道数据放在环形队列的那个位置了。

Sequence Barrier

Sequencer 生成一个 Sequence Barrier，其中包含对来自 Sequencer 的主要已发布Sequence和任何依赖消费者的Sequence的引用。

它的功能是确定是否有任何事件可供消费者处理。

Wait Strategy

Wait Strategy 决定了consumer如何等待producer把事件放入 Disruptor 中。

更多细节查看 lock-free 的内容。

Event

从producer传递到consumer的数据单元 (unit of data) 被称为 Event。

并非Disruptor指定类型，由用户自定义。

Event Processor

用于处理来自 Disruptor 的事件的主事件循环，并拥有消费者序列的所有权。有一个称为 BatchEventProcessor 的表示，它包含事件循环的有效实现，并将回调到使用过的 EventHandler 接口的提供实现。

The main event loop for handling events from the Disruptor and has ownership of consumer’s Sequence. There is a single representation called BatchEventProcessor that contains an efficient implementation of the event loop and will call back onto a used supplied implementation of the EventHandler interface.

Event Handler

由用户实现的 Disruptor 的消费者接口，用于处理事件，是 Consumer 的真正实现。

Producer

用户代码，调用Disruptor使生成的 Event s 进入队列。Disruptor没有定义特定接口或类型

Core Process

• Using the Disruptor
  • Introduction
  • Getting Started
  • Advanced Techniques
• Design and Implementation
• Known Issues
• Batch Rewind
  • The Feature
  • Use Case

Source

LMAX Disruptor User Guide (lmax-exchange.github.io)

比较 Kafka和disruptor？ - 知乎 (zhihu.com)

https://www.zhihu.com/question/323668956

https://www.zhihu.com/question/23235063

https://zhuanlan.zhihu.com/p/565937436

• start() && run() 源码
• 区别
• Demo及Demo遇到的问题 (Method Reference)

Thread

Source Code

Thread start()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

/**
     * Causes this thread to begin execution; the Java Virtual Machine
     * calls the <code>run</code> method of this thread.
     * <p>
     * The result is that two threads are running concurrently: the
     * current thread (which returns from the call to the
     * <code>start</code> method) and the other thread (which executes its
     * <code>run</code> method).
     * <p>
     * It is never legal to start a thread more than once.
     * In particular, a thread may not be restarted once it has completed
     * execution.
     *
     * @exception  IllegalThreadStateException  if the thread was already
     *               started.
     * @see        #run()
     * @see        #stop()
     */
public synchronized void start() {
  /**
         * This method is not invoked for the main method thread or "system"
         * group threads created/set up by the VM. Any new functionality added
         * to this method in the future may have to also be added to the VM.
         *
         * A zero status value corresponds to state "NEW".
         */
  if (threadStatus != 0)
    throw new IllegalThreadStateException();

  /* Notify the group that this thread is about to be started
         * so that it can be added to the group's list of threads
         * and the group's unstarted count can be decremented. */
  group.add(this);

  boolean started = false;
  try {
    start0();
    started = true;
  } finally {
    try {
      if (!started) {
        group.threadStartFailed(this);
      }
    } catch (Throwable ignore) {
      /* do nothing. If start0 threw a Throwable then
                  it will be passed up the call stack */
    }
  }
}

Thread run()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

/**
     * If this thread was constructed using a separate
     * <code>Runnable</code> run object, then that
     * <code>Runnable</code> object's <code>run</code> method is called;
     * otherwise, this method does nothing and returns.
     * <p>
     * Subclasses of <code>Thread</code> should override this method.
     *
     * @see     #start()
     * @see     #stop()
     * @see     #Thread(ThreadGroup, Runnable, String)
     */
@Override
public void run() {
  if (target != null) {
    target.run();
  }
}

Diff

start() 方法是真正的启动方法，是并行调用的

run()是Thread中的一个方法，是串行调用的

函数式接口