使用CycleBarrier实现异步等待

Posted on 2019-07-03 Edited on 2026-08-23 In Java

可循环利用的屏障
每个线程执行时，都会碰到一个屏障，直到所有线程执行结束，然后屏障便会打开，使所有线程继续往下执行

CyclicBarrier(int parties)和CyclicBarrier(int parties, Runnable barrierAction)

前者只需要声明需要拦截的线程数即可
后者还需要定义一个等待所有线程到达屏障优先执行的Runnable对象

实现原理：

在CyclicBarrier的内部定义了一个Lock对象，每当一个线程调用await方法时，将拦截的线程数减1，然后判断剩余拦截数是否为初始值parties，如果不是，进入Lock对象的条件队列等待。如果是，执行barrierAction对象的Runnable方法，然后将锁的条件队列中的所有线程放入锁等待队列中，这些线程会依次的获取锁、释放锁。

使用CountLatch实现异步等待

Posted on 2019-07-03 Edited on 2026-08-23 In Java

10.async-wait-with-latch

利用CountLatch异步等待

摘自JavaFx源码

com.sun.javafx.application.LauncherImpl#launchApplication(java.lang.Class<? extends javafx.application.Application>, java.lang.Class<? extends javafx.application.Preloader>, java.lang.String[])

// 接收异常
private static volatile RuntimeException launchException = null;


final CountDownLatch launchLatch = new CountDownLatch(1);
Thread launcherThread = new Thread(() -> {
    try {
        launchApplication1(appClass, preloaderClass, args);
    } catch (RuntimeException rte) {
        launchException = rte;
    } catch (Exception ex) {
        launchException =
            new RuntimeException("Application launch exception", ex);
    } catch (Error err) {
        launchException =
            new RuntimeException("Application launch error", err);
    } finally {
    // countDown
        launchLatch.countDown();
    }
});
launcherThread.setName("JavaFX-Launcher");
launcherThread.start();
// 异步线程启动

// Wait for FX launcher thread to finish before returning to user
try {
    launchLatch.await();
} catch (InterruptedException ex) {
    throw new RuntimeException("Unexpected exception: ", ex);
}

if (launchException != null) {
    throw launchException;
}

【转载】flink cep官方文档

Posted on 2019-06-09 Edited on 2026-08-23 In Distributed

转载自 https://github.com/crestofwave1/oneFlink/blob/master/doc/CEP/FlinkCEPOfficeWeb.md

0. 本文概述简介

FlinkCEP是在Flink之上实现的复杂事件处理（CEP）库。它允许你在无界的事件流中检测事件模式，让你有机会掌握数据中重要的事项。

本文描述了Flink CEP中可用的API调用。首先介绍Pattern API，它允许你指定要在流中检测的模式，然后介绍如何检测匹配事件序列并对其进行操作。然后，我们将介绍CEP库在处理事件时间延迟时所做的假设。

1.入门

首先是要在你的pom.xml文件中，引入CEP库。

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-cep_2.11</artifactId>
  <version>1.5.0</version>
</dependency>

注意要应用模式匹配的DataStream中的事件必须实现正确的equals（）和hashCode（）方法，因为FlinkCEP使用它们来比较和匹配事件。

第一个demo如下：

DataStream<Event> input = ...

Pattern<Event, ?> pattern = Pattern.<Event>begin("start").where(
        new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) {
                return event.getId() == 42;
            }
        }
    ).next("middle").subtype(SubEvent.class).where(
        new SimpleCondition<Event>() {
            @Override
            public boolean filter(SubEvent subEvent) {
                return subEvent.getVolume() >= 10.0;
            }
        }
    ).followedBy("end").where(
         new SimpleCondition<Event>() {
            @Override
            public boolean filter(Event event) {
                return event.getName().equals("end");
            }
         }
    );

PatternStream<Event> patternStream = CEP.pattern(input, pattern);

DataStream<Alert> result = patternStream.select(
    new PatternSelectFunction<Event, Alert> {
        @Override
        public Alert select(Map<String, List<Event>> pattern) throws Exception {
            return createAlertFrom(pattern);
        }
    }
});

2.Pattern API

Pattern API允许你定义要从输入流中提取的复杂模式序列。

每个复杂模式序列都是由多个简单模式组成，即寻找具有相同属性的单个事件的模式。我们可以先定义一些简单的模式，然后组合成复杂的模式序列。可以将模式序列视为此类模式的结构图，基于用户指定的条件从一个模式转换到下一个模式，例如， event.getName().equals(“start”)。匹配是一系列输入事件，通过一系列有效的模式转换访问复杂模式图中的所有模式。

注意每个模式必须具有唯一的名称，以便后续可以使用该名称来标识匹配的事件。

注意模式名称不能包含字符“：”。

在本节接下来的部分，我们将首先介绍如何定义单个模式，然后如何将各个模式组合到复杂模式中。

2.1 单个模式

Pattern可以是单单个，也可以是循环模式。单个模式接受单个事件，而循环模式可以接受多个事件。在模式匹配符号中，模式“a b + c？d”（或“a”，后跟一个或多个“b”，可选地后跟“c”，后跟“d”），a，c ？，和d是单例模式，而b +是循环模式。默认情况下，模式是单个模式，您可以使用Quantifiers将其转换为循环模式。每个模式可以有一个或多个条件，基于它接受事件。

2.1.1 Quantifiers

在FlinkCEP中，您可以使用以下方法指定循环模式：pattern.oneOrMore（），用于期望一个或多个事件发生的模式（例如之前提到的b +）;和pattern.times（#ofTimes），用于期望给定类型事件的特定出现次数的模式，例如4个;和patterntimes（#fromTimes，＃toTimes），用于期望给定类型事件的最小出现次数和最大出现次数的模式，例如， 2-4。

您可以使用pattern.greedy（）方法使循环模式变得贪婪，但是还不能使组模式变得贪婪。您可以使用pattern.optional（）方法使得所有模式，循环与否，变为可选。

对于名为start的模式，以下是有效的Quantifiers：

// expecting 4 occurrences
start.times(4);

// expecting 0 or 4 occurrences
start.times(4).optional();

// expecting 2, 3 or 4 occurrences
start.times(2, 4);

// expecting 2, 3 or 4 occurrences and repeating as many as possible
start.times(2, 4).greedy();

// expecting 0, 2, 3 or 4 occurrences
start.times(2, 4).optional();

// expecting 0, 2, 3 or 4 occurrences and repeating as many as possible
start.times(2, 4).optional().greedy();

// expecting 1 or more occurrences
start.oneOrMore();

// expecting 1 or more occurrences and repeating as many as possible
start.oneOrMore().greedy();

// expecting 0 or more occurrences
start.oneOrMore().optional();

// expecting 0 or more occurrences and repeating as many as possible
start.oneOrMore().optional().greedy();

// expecting 2 or more occurrences
start.timesOrMore(2);

// expecting 2 or more occurrences and repeating as many as possible
start.timesOrMore(2).greedy();

// expecting 0, 2 or more occurrences and repeating as many as possible
start.timesOrMore(2).optional().greedy();

2.1.2 Conditions-条件

在每个模式中，从一个模式转到下一个模式，可以指定其他条件。您可以将使用下面这些条件：

传入事件的属性，例如其值应大于5，或大于先前接受的事件的平均值。
匹配事件的连续性，例如检测模式a，b，c，序列中间不能有任何非匹配事件。

2.1.3 Conditions on Properties-关于属性的条件

可以通过pattern.where（），pattern.or（）或pattern.until（）方法指定事件属性的条件。条件可以是IterativeConditions或SimpleConditions。

迭代条件：

这是最常见的条件类型。你可以指定一个条件，该条件基于先前接受的事件的属性或其子集的统计信息来接受后续事件。

下面代码说的是：如果名称以“foo”开头同时如果该模式的先前接受的事件的价格总和加上当前事件的价格不超过该值 5.0，则迭代条件接受名为“middle”的模式的下一个事件，。迭代条件可以很强大的，尤其是与循环模式相结合，例如， oneOrMore()。

middle.oneOrMore().where(new IterativeCondition<SubEvent>() {
    @Override
    public boolean filter(SubEvent value, Context<SubEvent> ctx) throws Exception {
        if (!value.getName().startsWith("foo")) {
            return false;
        }

        double sum = value.getPrice();
        for (Event event : ctx.getEventsForPattern("middle")) {
            sum += event.getPrice();
        }
        return Double.compare(sum, 5.0) < 0;
    }
});

注意对context.getEventsForPattern（…）的调用,将为给定潜在匹配项查找所有先前接受的事件。此操作的代价可能会变化巨大，因此在使用条件时，请尽量减少其使用。

简单条件：

这种类型的条件扩展了前面提到的IterativeCondition类，并且仅根据事件本身的属性决定是否接受事件。

start.where(new SimpleCondition<Event>() {
    @Override
    public boolean filter(Event value) {
        return value.getName().startsWith("foo");
    }
});

最后，还可以通过pattern.subtype（subClass）方法将接受事件的类型限制为初始事件类型的子类型。

start.subtype(SubEvent.class).where(new SimpleCondition<SubEvent>() {
    @Override
    public boolean filter(SubEvent value) {
        return ... // some condition
    }
});

组合条件：

如上所示，可以将子类型条件与其他条件组合使用。这适用于所有条件。您可以通过顺序调用where（）来任意组合条件。最终结果将是各个条件的结果的逻辑AND。要使用OR组合条件，可以使用or（）方法，如下所示。

pattern.where(new SimpleCondition<Event>() {
    @Override
    public boolean filter(Event value) {
        return ... // some condition
    }
}).or(new SimpleCondition<Event>() {
    @Override
    public boolean filter(Event value) {
        return ... // or condition
    }
});

停止条件：

在循环模式（oneOrMore()和oneOrMore().optional()）的情况下，还可以指定停止条件，例如：接受值大于5的事件，直到值的总和小于50。

为了更好的理解，可以看看下面的例子：

给定模式：(a+ until b)，b之前，要出现一个或者多个a。

给定输入序列：a1，c，a2，b，a3

输出结果: {a1 a2}{a1}{a2}{a3}

可以看到{a1,a2,a3},{a2,a3}这两个并没有输出，这就是停止条件的作用。

连续事件条件

FlinkCEP支持事件之间以下形式进行连续：

严格连续性：希望所有匹配事件一个接一个地出现，中间没有任何不匹配的事件。
宽松连续性：忽略匹配的事件之间出现的不匹配事件。不能忽略两个事件之间的匹配事件。
非确定性轻松连续性：进一步放宽连续性，允许忽略某些匹配事件的其他匹配。

为了解释上面的内容，我们举个例子。假如有个模式序列”a+ b”，输入序列”a1,c,a2,b”，不同连续条件下有不同的区别：

严格连续性：{a2 b} - 由于c的存在导致a1被废弃
宽松连续性：{a1,b}和{a1 a2 b} - c被忽略
非确定性宽松连续性：{a1 b}, {a2 b}, 和 {a1 a2 b}

对于循环模式（例如oneOrMore()和times()），默认是宽松的连续性。如果你想要严格的连续性，你必须使用consecutive()显式指定它，如果你想要非确定性的松弛连续性，你可以使用allowCombinations()方法。

注意在本节中，我们讨论的是单个循环模式中的连续性，并且需要在该上下文中理解consecutive()和allowCombinations()。稍后在讲解组合模式时，我们将讨论其他方法，例如next（）和followedBy（），用于指定模式之间的连续条件。

2.1.4 API简介

1. where(condition)

定义当前模式的条件。为了匹配模式，事件必须满足条件。多个连续的where()，其条件为AND：

pattern.where(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ... // some condition
    }
});

2. or(condition)

添加与现有条件进行OR运算的新条件。只有在至少通过其中一个条件时，事件才能匹配该模式：

pattern.where(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ... // some condition
    }
}).or(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ... // alternative condition
    }
});

3. until(condition)

指定循环模式的停止条件。意味着如果匹配给定条件的事件发生，则不再接受该模式中的事件。

仅适用于oneOrMore（）

注意：它允许在基于事件的条件下清除相应模式的状态。

pattern.oneOrMore().until(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ... // alternative condition
    }
});

4. subtype(subClass)

定义当前模式的子类型条件。如果事件属于此子类型，则事件只能匹配该模式：

1	pattern.subtype(SubEvent.class);

5. oneOrMore()

指定此模式至少发生一次匹配事件。

默认情况下，使用宽松的内部连续性。

注意：建议使用until（）或within（）来启用状态清除

pattern.oneOrMore().until(new IterativeCondition<Event>() {
    @Override
    public boolean filter(Event value, Context ctx) throws Exception {
        return ... // alternative condition
    }
});

6. timesOrMore(#times)

指定此模式至少需要#times次出现匹配事件。

默认情况下，使用宽松的内部连续性（在后续事件之间）。

1	pattern.timesOrMore(2);

7. times(#ofTimes)

指定此模式需要匹配事件的确切出现次数。

默认情况下，使用宽松的内部连续性（在后续事件之间）。

1	pattern.times(2);

8. times(fromTimes, toTimes)

指定此模式期望在匹配事件的#fromTimes次和#toTimes次之间出现。

默认情况下，使用宽松的内部连续性。

1	pattern.times(2, 4);

9. optional()

指定此模式是可选的，即有可能根本不会发生。这适用于所有上述量词。

1	pattern.oneOrMore().optional();

10. greedy()

指定此模式是贪婪的，即它将尽可能多地重复。这仅适用于quantifiers，目前不支持组模式。

1	pattern.oneOrMore().greedy();

11. consecutive()

与oneOrMore（）和times（）一起使用并在匹配事件之间强加严格的连续性，即任何不匹配的元素都会中断匹配。

如果不使用，则使用宽松的连续性（如followBy（））。

例如，这样的模式：

Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
  @Override
  public boolean filter(Event value) throws Exception {
    return value.getName().equals("c");
  }
})
.followedBy("middle").where(new SimpleCondition<Event>() {
  @Override
  public boolean filter(Event value) throws Exception {
    return value.getName().equals("a");
  }
}).oneOrMore().consecutive()
.followedBy("end1").where(new SimpleCondition<Event>() {
  @Override
  public boolean filter(Event value) throws Exception {
    return value.getName().equals("b");
  }
});

针对上面的模式，我们假如输入序列如：C D A1 A2 A3 D A4 B

使用consecutive：{C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B}

不使用:{C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B}, {C A1 A2 A3 A4 B}

12. allowCombinations()

与oneOrMore（）和times（）一起使用，并在匹配事件之间强加非确定性宽松连续性（如 followedByAny()）。

如果不应用，则使用宽松的连续性（如followBy()）。

例如,这样的模式：

Pattern.<Event>begin("start").where(new SimpleCondition<Event>() {
  @Override
  public boolean filter(Event value) throws Exception {
    return value.getName().equals("c");
  }
})
.followedBy("middle").where(new SimpleCondition<Event>() {
  @Override
  public boolean filter(Event value) throws Exception {
    return value.getName().equals("a");
  }
}).oneOrMore().allowCombinations()
.followedBy("end1").where(new SimpleCondition<Event>() {
  @Override
  public boolean filter(Event value) throws Exception {
    return value.getName().equals("b");
  }
});

针对上面的模式，我们假如输入序列如：C D A1 A2 A3 D A4 B

使用allowCombinations：{C A1 B}, {C A1 A2 B}, {C A1 A3 B}, {C A1 A4 B}, {C A1 A2 A3 B}, {C A1 A2 A4 B}, {C A1 A3 A4 B}, {C A1 A2 A3 A4 B}

不使用:{C A1 B}, {C A1 A2 B}, {C A1 A2 A3 B}, {C A1 A2 A3 A4 B}

2.2 组合模式

2.2.1 简介

已经了解了单个模式的样子，现在是时候看看如何将它们组合成一个完整的模式序列。

模式序列必须以初始模式开始，如下所示：

1	Pattern<Event, ?> start = Pattern.<Event>begin("start");

接下来，您可以通过指定它们之间所需的连续条件，为模式序列添加更多模式。在上一节中，我们描述了Flink支持的不同邻接模式，即严格，宽松和非确定性宽松，以及如何在循环模式中应用它们。要在连续模式之间应用它们，可以使用：

next() 对应严格, followedBy() 对应宽松连续性 followedByAny() 对应非确定性宽松连续性

亦或

notNext() 如果不希望一个事件类型紧接着另一个类型出现。 notFollowedBy() 不希望两个事件之间任何地方出现该事件。

注意模式序列不能以notFollowedBy（）结束。

注意 NOT模式前面不能有可选模式。

// strict contiguity
Pattern<Event, ?> strict = start.next("middle").where(...);

// relaxed contiguity
Pattern<Event, ?> relaxed = start.followedBy("middle").where(...);

// non-deterministic relaxed contiguity
Pattern<Event, ?> nonDetermin = start.followedByAny("middle").where(...);

// NOT pattern with strict contiguity
Pattern<Event, ?> strictNot = start.notNext("not").where(...);

// NOT pattern with relaxed contiguity
Pattern<Event, ?> relaxedNot = start.notFollowedBy("not").where(...);

宽松连续性指的是仅第一个成功匹配的事件会被匹配到，然而非确定性宽松连续性，相同的开始会有多个匹配结果发出。距离，如果一个模式是”a b”，给定输入序列是”a c b1 b2”。对于不同连续性会有不同输出。

a和b之间严格连续性，将会返回{},也即是没有匹配。因为c的出现导致a，抛弃了。
a和b之间宽松连续性，返回的是{a，b1},因为宽松连续性将会抛弃为匹配成功的元素，直至匹配到下一个要匹配的事件。
a和b之间非确定性宽松连续性，返回的是{a,b1},{a,b2}。

也可以为模式定义时间约束。例如，可以通过pattern.within（）方法定义模式应在10秒内发生。时间模式支持处理和事件时间。注意模式序列只能有一个时间约束。如果在不同的单独模式上定义了多个这样的约束，则应用最小的约束。

1	next.within(Time.seconds(10));

可以为begin，followBy，followByAny和next定义一个模式序列作为条件。模式序列将被逻辑地视为匹配条件，而且将返回GroupPattern并且可对GroupPattern使用oneOrMore（），times（#ofTimes），times（＃fromTimes，＃toTimes），optional（），consecutive（）， allowCombinations（）等方法。

PatternPatte <Event, ?> start = Pattern.begin(
    Pattern.<Event>begin("start").where(...).followedBy("start_middle").where(...)
);

// strict contiguity
Pattern<Event, ?> strict = start.next(
    Pattern.<Event>begin("next_start").where(...).followedBy("next_middle").where(...)
).times(3);

// relaxed contiguity
Pattern<Event, ?> relaxed = start.followedBy(
    Pattern.<Event>begin("followedby_start").where(...).followedBy("followedby_middle").where(...)
).oneOrMore();

// non-deterministic relaxed contiguity
Pattern<Event, ?> nonDetermin = start.followedByAny(
    Pattern.<Event>begin("followedbyany_start").where(...).followedBy("followedbyany_middle").where(...)
).optional();

2.2.2 API

begin(#name)

定义一个开始模式

1	Pattern<Event, ?> start = Pattern.<Event>begin("start");

begin(#pattern_sequence)

定义一个开始模式

1
2
3

Pattern<Event, ?> start = Pattern.<Event>begin(
    Pattern.<Event>begin("start").where(...).followedBy("middle").where(...)
);

next(#name)

追加一个新的模式。匹配事件必须直接跟着先前的匹配事件（严格连续性）：

1	Pattern<Event, ?> next = start.next("middle");

next(#pattern_sequence)

追加一个新的模式。匹配事件必须直接接着先前的匹配事件（严格连续性）：

1
2
3

Pattern<Event, ?> next = start.next(
    Pattern.<Event>begin("start").where(...).followedBy("middle").where(...)
);

followedBy(#name)

追加加新模式。匹配事件和先前匹配事件（宽松连续）之间可能发生其他非匹配事件：

1	Pattern<Event, ?> followedBy = start.followedBy("middle");

followedBy(#pattern_sequence)

追加新模式。匹配事件和先前匹配事件（宽松连续）之间可能发生其他非匹配事件：

1
2
3

Pattern<Event, ?> followedBy = start.followedBy(
    Pattern.<Event>begin("start").where(...).followedBy("middle").where(...)
);

followedByAny(#name)

添加新模式。匹配事件和先前匹配事件之间可能发生其他事件，并且将针对每个备选匹配事件（非确定性放松连续性）呈现替代匹配：

1	Pattern<Event, ?> followedByAny = start.followedByAny("middle");

followedByAny(#pattern_sequence)

添加新模式。匹配事件和先前匹配事件之间可能发生其他事件，并且将针对每个备选匹配事件（非确定性放松连续性）呈现替代匹配：

1
2
3

Pattern<Event, ?> followedByAny = start.followedByAny(
    Pattern.<Event>begin("start").where(...).followedBy("middle").where(...)
);

notNext()

添加新的否定模式。匹配（否定）事件必须直接跟着先前的匹配事件（严格连续性）才能丢弃部分匹配：

1	Pattern<Event, ?> notNext = start.notNext("not");

notFollowedBy()

追加一个新的否定模式匹配。即使在匹配（否定）事件和先前匹配事件（宽松连续性）之间发生其他事件，也将丢弃部分匹配事件序列：

1	Pattern<Event, ?> notFollowedBy = start.notFollowedBy("not");

within(time)

定义事件序列进行模式匹配的最大时间间隔。如果未完成的事件序列超过此时间，则将其丢弃：

1	pattern.within(Time.seconds(10));

2.3 匹配后的跳过策略

对于给定模式，可以将同一事件分配给多个成功匹配。要控制将分配事件的匹配数，需要指定名为AfterMatchSkipStrategy的跳过策略。跳过策略有四种类型，如下所示：

NO_SKIP：将发出每个可能的匹配。
SKIP_PAST_LAST_EVENT：丢弃包含匹配事件的每个部分匹配。
SKIP_TO_FIRST：丢弃包含PatternName第一个之前匹配事件的每个部分匹配。
SKIP_TO_LAST：丢弃包含PatternName最后一个匹配事件之前的每个部分匹配。

请注意，使用SKIP_TO_FIRST和SKIP_TO_LAST跳过策略时，还应指定有效的PatternName。

例如，对于给定模式a b {2}和数据流ab1，ab2，ab3，ab4，ab5，ab6，这四种跳过策略之间的差异如下：

要指定要使用的跳过策略，只需调用以下命令创建AfterMatchSkipStrategy：

使用方法：

1 2	AfterMatchSkipStrategy skipStrategy = ... Pattern.begin("patternName", skipStrategy);

2.4 检测模式-Detecting Patterns

指定要查找的模式序列后，就可以将其应用于输入流以检测潜在匹配。要针对模式序列运行事件流，必须创建PatternStream。给定输入流 input，模式 pattern 和可选的比较器 comparator，用于在EventTime的情况下对具有相同时间戳的事件进行排序或在同一时刻到达，通过调用以下命令创建PatternStream：

DataStream<Event> input = ...
Pattern<Event, ?> pattern = ...
EventComparator<Event> comparator = ... // optional

PatternStream<Event> patternStream = CEP.pattern(input, pattern, comparator);

根据实际情况，创建的流可以是有key，也可以是无key的。

请注意，在无key的流上使用模式，将导致job的并行度为1。

2.5 Selecting from Patterns

获得PatternStream后，您可以通过select或flatSelect方法从检测到的事件序列中进行查询。

select（）方法需要PatternSelectFunction的实现。 PatternSelectFunction具有为每个匹配事件序列调用的select方法。它以Map <String，List >的形式接收匹配，其中key是模式序列中每个模式的名称，值是该模式的所有已接受事件的列表（IN是输入元素的类型）。给定模式的事件按时间戳排序。返回每个模式的接受事件列表的原因是当使用循环模式（例如oneToMany（）和times（））时，对于给定模式可以接受多个事件。选择函数只返回一个结果。

class MyPatternSelectFunction<IN, OUT> implements PatternSelectFunction<IN, OUT> {
    @Override
    public OUT select(Map<String, List<IN>> pattern) {
        IN startEvent = pattern.get("start").get(0);
        IN endEvent = pattern.get("end").get(0);
        return new OUT(startEvent, endEvent);
    }
}

PatternFlatSelectFunction类似于PatternSelectFunction，唯一的区别是它可以返回任意数量的结果。为此，select方法有一个额外的Collector参数，用于将输出元素向下游转发。

class MyPatternFlatSelectFunction<IN, OUT> implements PatternFlatSelectFunction<IN, OUT> {
    @Override
    public void flatSelect(Map<String, List<IN>> pattern, Collector<OUT> collector) {
        IN startEvent = pattern.get("start").get(0);
        IN endEvent = pattern.get("end").get(0);

        for (int i = 0; i < startEvent.getValue(); i++ ) {
            collector.collect(new OUT(startEvent, endEvent));
        }
    }
}

2.6 处理超时部分模式

每当模式具有通过within关键字附加的时间窗口长度时，部分事件序列可能因为超出时间窗口长度而被丢弃。为了对这些超时的部分匹配作出相应的处理，select和flatSelect API调用允许指定超时处理程序。为每个超时的部分事件序列调用此超时处理程序。超时处理程序接收到目前为止由模式匹配的所有事件，以及检测到超时时的时间戳。

为了处理部分模式，select和flatSelect API提供了一个带参数的重载版本

PatternTimeoutFunction/ PatternFlatTimeoutFunction。
OutputTag 超时的匹配将会在其中返回。
PatternSelectFunction / PatternFlatSelectFunction。

PatternStreamPatte <Event> patternStream = CEP.pattern(input, pattern);

OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<ComplexEvent> result = patternStream.select(
    new PatternTimeoutFunction<Event, TimeoutEvent>() {...},
    outputTag,
    new PatternSelectFunction<Event, ComplexEvent>() {...}
);

DataStream<TimeoutEvent> timeoutResult = result.getSideOutput(outputTag);

SingleOutputStreamOperator<ComplexEvent> flatResult = patternStream.flatSelect(
    new PatternFlatTimeoutFunction<Event, TimeoutEvent>() {...},
    outputTag,
    new PatternFlatSelectFunction<Event, ComplexEvent>() {...}
);

DataStream<TimeoutEvent> timeoutFlatResult = flatResult.getSideOutput(outputTag);

2.7 事件事件模式下处理滞后数据

在CEP中，元素处理的顺序很重要。为了保证在采用事件事件时以正确的顺序处理事件，最初将传入的事件放入缓冲区，其中事件基于它们的时间戳以升序排序，并且当watermark到达时，处理该缓冲区中时间戳小于watermark时间的所有元素。这意味着watermark之间的事件按事件时间顺序处理。

请注意，在采用事件时间时，CEP library会假设watermark是正确的。

为了保证跨watermark的记录按照事件事件顺序处理，Flink的CEP库假定watermark是正确的，并将时间戳小于上次可见watermark的时间视为滞后事件。滞后事件不会被进一步处理。

2.8 栗子

以下示例检测事件的带key数据流上的模式start，middle（name =“error”） - > end（name =“critical”）。事件的key是其id，并且有效模式必须在10秒内发生。整个处理是用事件时间完成的。

StreamExecutionEnvironment env = ...
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

DataStream<Event> input = ...

DataStream<Event> partitionedInput = input.keyBy(new KeySelector<Event, Integer>() {
	@Override
	public Integer getKey(Event value) throws Exception {
		return value.getId();
	}
});

Pattern<Event, ?> pattern = Pattern.<Event>begin("start")
	.next("middle").where(new SimpleCondition<Event>() {
		@Override
		public boolean filter(Event value) throws Exception {
			return value.getName().equals("error");
		}
	}).followedBy("end").where(new SimpleCondition<Event>() {
		@Override
		public boolean filter(Event value) throws Exception {
			return value.getName().equals("critical");
		}
	}).within(Time.seconds(10));

PatternStream<Event> patternStream = CEP.pattern(partitionedInput, pattern);

DataStream<Alert> alerts = patternStream.select(new PatternSelectFunction<Event, Alert>() {
	@Override
	public Alert select(Map<String, List<Event>> pattern) throws Exception {
		return createAlert(pattern);
	}
});

Flink/Storm 动态更新配置实现方案

Posted on 2019-06-09 Edited on 2026-08-23 In Distributed

实时计算处理无限数据流，对系统可用性十分敏感，然而业务需求具有必然的更新需求，动态更新实时计算的配置是常见的需求，比如动态增加用户白名单、业务数据在线 debug、新增广告位统计等等。然而，实现并不简单，Apache Strom 和 Apache Storm 具有不同的架构，实现方式也不尽相同。

Apache Strom(这里不包括 Trident)，所有的计算逻辑都是通过实现spout 和 bolt，运行在 task节点上，因此与业务逻辑相关的状态管理、配置管理以及输入输出的控制等均可以定义在 bolt 中，如定时一分钟轮询配置变化、定时 checkpoint 近似数据集等。此外，配置同样可以通过控制流的方式与数据流 uion 到一起，在计算节点检测判断控制流的关键字来实现配置更新的目的。

Apache Flink 的函数式 API 封装要比 Storm 要完善，而灵活性不比 Strom。实现配置动态更新有两种方式，一是在算子中定时轮询拉取配置信息，二是利用广播状态和控制流。

asyncIO

stateBroadcast

[思考] 实时数仓

Posted on 2019-06-06 Edited on 2026-08-23 In Distributed

实时数仓

为什么构建实时数仓

[转载]基于跳跃表的 ConcurrentSkipListMap 内部实现（Java 8）

Posted on 2019-06-06 Edited on 2026-08-23 In Flink

基于跳跃表的 ConcurrentSkipListMap 内部实现（Java 8）

我们知道 HashMap 是一种键值对形式的数据存储容器，但是它有一个缺点是，元素内部无序。由于它内部根据键的 hash 值取模表容量来得到元素的存储位置，所以整体上说 HashMap 是无序的一种容器。当然，jdk 中也为我们提供了基于红黑树的存储的 TreeMap 容器，它的内部元素是有序的，但是由于它内部通过红黑结点的各种变换来维持二叉搜索树的平衡，相对复杂，并且在并发环境下碍于 rebalance 操作，性能会受到一定的影响。

跳表（SkipList）是一种随机化的数据结构，通过“空间来换取时间”的一个算法，建立多级索引，实现以二分查找遍历一个有序链表。时间复杂度等同于红黑树，O(log n)。但实现却远远比红黑树要简单，本篇我们主要从以下几个方面来对这种并发版本的数据结构进行学习：

跳跃表的数据结构介绍
ConcurrentSkipListMap 的前导知识预备
基本的成员属性介绍
put 方法并发添加
remove 方法的并发删除
get 方法获取指定结点的 value
其它的一些方法的简单描述

一、跳跃表的数据结构介绍

跳跃表具有以下几个必备的性质：

最底层包含所有节点的一个有序的链表
每一层都是一个有序的链表
每个节点都有两个指针，一个指向右侧节点（没有则为空），一个指向下层节点（没有则为空）
必备一个头节点指向最高层的第一个节点，通过它可以遍历整张表

当我们查找一个元素的时候就是这样的：

查找的过程有点像我们的二分查找，不过这里我们是通过为链表建立多级索引，以空间换时间来实现二分查找。所以，跳表的查询操作的时间复杂度为 O(logN)。

接着我们看看跳表的插入操作：
首先，跳表的插入必然会在底层增加一个节点，但是往上的层次是否需要增加节点则完全是随机的，SkipList 通过概率保证整张表的节点分布均匀，它不像红黑树是通过人为的 rebalance 操作来保证二叉树的平衡性。（数学对于计算机还是很重要的）。

通过概率算法得到新插入节点的一个 level 值，如果小于当前表的最大 level，从最底层到 level 层都添加一个该节点。例如：

如图，首先 119 节点会被添加到最底层链表的合适位置，然后通过概率算法得到 level 为 2，于是 1—level 层中的每一层都添加了 119 节点。

如果概率算法得到的 level 大于当前表的最大 level 值的话，那么将会新增一个 level，并且将新节点添加到该 level 上。

跳表的删除操作其实就是一个查找加删除节点的操作

好了，有关跳表这种数据结构的基本理论知识已经简单的介绍了，下面我们看 jdk 中对该数据结构的基本实现情况，并了解它的并发版本是如何实现的。

二、ConcurrentSkipListMap 的前导知识预备

在实际分析 put 方法之前，有一些预备的知识我们需要先有个大致的了解，否则在实际分析源码的时候会感觉吃力些。

首先是删除操作，在我们上述的跳表数据结构中谈及的删除操作主要是定位待删结点+删除该结点的一个复合操作。而在我们的并发跳表中，删除操作相对复杂点，需要分为以下三个步骤：

找到待删结点并将其 value 属性值由 notnull 置为 null，整个过程是基于 CAS 无锁式算法的
向待删结点的 next 位置新增一个 marker 标记结点，整个过程也是基于 CAS 无锁式算法
CAS 式删除具体的结点，实际上也就是跳过该待删结点，让待删结点的前驱节点直接越过本身指向待删结点的后继结点即可

例如我们有以下三个结点，n 为待删除的结点。

+------+ +------+ +——+
… | b |——>| n |—–>| f | …
+------+ +------+ +——+

第一步是找到 n ，然后 CAS 该结点的 value 值为 null。如果该步骤失败了，那么 ConcurrentSkipListMap 会通过循环再次尝试 CAS 将 n 的 value 属性赋值为 null。

第二步是建立在第一步成功的前提下的，n 的当前 value 属性的值为 null，ConcurrentSkipListMap 试图在 n 的后面增加一个空的 node 结点（marker）以分散下一步的并发冲突性。

+------+ +------+ +------+ +——+
… | b |——>| n |—–>|marker|—->| f | …
+------+ +------+ +------+ +——+

第三步，断链操作。如果 marker 添加失败，将不会有第三步，直接回重新回到第一步。如果成功添加，那么将试图断开 b 到 n 的链接，直接绕过 n，让 b 的 next 指向 f。那么，这个 n 结点将作为内存中的一个游离结点，最终被 GC 掉。断开失败的话，也将回到第一步。

+------+ +——+
… | b |———————–>| f | …
+------+ +——+

主要还是有关删除这方面的预备知识，其它的信息点我们将从实际方法的源码中再进行分析。

三、基本的成员属性介绍

static final class Node<K,V> {
    final K key;
    volatile Object value;
    volatile Node<K,V> next;

    Node(K key, Object value, Node<K,V> next) {
        this.key = key;
        this.value = value;
        this.next = next;
    }
    //省略其它的一些基于当前结点的 CAS 方法
}

这是 node 结点类型的定义，是最基本的数据存储单元。

static class Index<K,V> {
    final Node<K,V> node;
    final Index<K,V> down;
    volatile Index<K,V> right;

    Index(Node<K,V> node, Index<K,V> down, Index<K,V> right) {
        this.node = node;
        this.down = down;
        this.right = right;
    }
    //省略其它的一些基于当前结点的 CAS 方法
}

Index 结点封装了 node 结点，作为跳表的最基本组成单元。

static final class HeadIndex<K,V> extends Index<K,V> {
    final int level;
    HeadIndex(Node<K,V> node, Index<K,V> down, Index<K,V> right, int level) {
        super(node, down, right);
        this.level = level;
    }
}

封装了 Index 结点，作为每层的头结点，level 属性用于标识当前层次的序号。

/**
 * The topmost head index of the skiplist.
 */
private transient volatile HeadIndex<K,V> head;

整个跳表的头结点，通过它可以遍历访问整张跳表。

1
2
3

//比较器，用于比较两个元素的键值大小，如果没有显式传入则默认为自然排序
final Comparator<? super K> comparator;

/**
 * Special value used to identify base-level header
 * 特殊的值，用于初始化跳表
 */
private static final Object BASE_HEADER = new Object();

紧接着，我们看看它的几个构造器：

//未传入比较器，则为默认值
public ConcurrentSkipListMap() {
    this.comparator = null;
    initialize();
}
public ConcurrentSkipListMap(Comparator<? super K> comparator) {
    this.comparator = comparator;
    initialize();
}

//所有的构造器都会调用这个初始化的方法
private void initialize() {
    keySet = null;
    entrySet = null;
    values = null;
    descendingMap = null;
    head = new HeadIndex<K,V>(new Node<K,V>(null, BASE_HEADER, null),null, null, 1);
}

这个初始化方法主要完成的是对整张跳表的一个初始化操作，head 头指针指向这个并没有什么实际意义的头结点。

基本的成员属性就简单介绍到这，重点还是那三个内部类，都分别代表了什么样的结点类型，都使用在何种场景下，务必清晰。

四、put 并发添加的内部实现

//基本的 put 方法，向跳表中添加一个节点
public V put(K key, V value) {
    if (value == null)
        throw new NullPointerException();
    return doPut(key, value, false);
}

put 方法的内部调用的是 doPut 方法来实现添加元素的，但是由于 doPut 方法的方法体很长，我们分几个部分进行分析。

//第一部分
private V doPut(K key, V value, boolean onlyIfAbsent) {
    Node<K,V> z;
    //边界值判断，空的 key 自然是不允许插入的
    if (key == null)
        throw new NullPointerException();
    //拿到比较器的引用
    Comparator<? super K> cmp = comparator;
    outer: for (;;) {
	    //根据 key，找到待插入的位置
	    //b 叫做前驱节点，将来作为新加入结点的前驱节点
	    //n 叫做后继结点，将来作为新加入结点的后继结点
	    //也就是说，新节点将插入在 b 和 n 之间
        for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
	        //如果 n 为 null，那么说明 b 是链表的最尾端的结点，这种情况比较简单，直接构建新节点插入即可
	        //否则走下面的判断体
            if (n != null) {
                Object v; int c;
                Node<K,V> f = n.next;
                //如果 n 不再是 b 的后继结点了，说明有其他线程向 b 后面添加了新元素
                //那么我们直接退出内循环，重新计算新节点将要插入的位置
                if (n != b.next)
                    break;
                //value =0 说明 n 已经被标识位待删除，其他线程正在进行删除操作
                //调用 helpDelete 帮助删除，并退出内层循环重新计算待插入位置
                if ((v = n.value) == null) { 
                    n.helpDelete(b, f);
                    break;
                }
                //b 已经被标记为待删除，前途结点 b 都丢了，可不得重新计算待插入位置吗
                if (b.value == null || v == n) 
                    break;
				//如果新节点的 key 大于 n 的 key 说明找到的前驱节点有误，按序往后挪一个位置即可
				//回到内层循环重新试图插入
                if ((c = cpr(cmp, key, n.key)) > 0) {
                    b = n;
                    n = f;
                    continue;
                }
                //新节点的 key 等于 n 的 key，这是一次 update 操作，CAS 更新即可
                //如果更新失败，重新进循环再来一次
                if (c == 0) {
                    if (onlyIfAbsent || n.casValue(v, value)) {
                        @SuppressWarnings("unchecked") V vv = (V)v;
                        return vv;
                    }
                    break; 
                }
            }
	    //无论遇到何种问题，到这一步说明待插位置已经确定
            z = new Node<K,V>(key, value, n);
            if (!b.casNext(n, z))
                break;
            //如果成功了，退出最外层循环，完成了底层的插入工作        
            break outer;
        }
    }

以上这一部分主要完成了向底层链表插入一个节点，至于其中具体的怎么找前驱节点的方法稍后介绍。但这其实只不过才完成一小半的工作，就像红黑树在插入后需要 rebalance 一样，我们的跳表需要根据概率算法保证节点分布稳定，它的调节措施相对于红黑树来说就简单多了，通过往上层索引层添加相关引用即可，以空间换时间。具体的我们来看：

//第二部分
//获取一个线程无关的随机数，占四个字节，32 个比特位
int rnd = ThreadLocalRandom.nextSecondarySeed();
	//和 1000 0000 0000 0000 0000 0000 0000 0001 与
	//如果等于 0，说明这个随机数最高位和最低位都为 0，这种概率很大
	//如果不等于 0，那么将仅仅把新节点插入到最底层的链表中即可，不会往上层递归
    if ((rnd & 0x80000001) == 0) { 
        int level = 1, max;
        //用低位连续为 1 的个数作为 level 的值，也是一种概率策略
        while (((rnd >>>= 1) & 1) != 0)
            ++level;
        Index<K,V> idx = null;
        HeadIndex<K,V> h = head;
        //如果概率算得的 level 在当前跳表 level 范围内
        //构建一个从 1 到 level 的纵列 index 结点引用
        if (level <= (max = h.level)) {
            for (int i = 1; i <= level; ++i)
                idx = new Index<K,V>(z, idx, null);
        }
        //否则需要新增一个 level 层
        else { 
            level = max + 1; 
            @SuppressWarnings("unchecked")
            Index<K,V>[] idxs =(Index<K,V>[])new Index<?,?>[level+1];
            for (int i = 1; i <= level; ++i)
                idxs[i] = idx = new Index<K,V>(z, idx, null);
            for (;;) {
                h = head;
                int oldLevel = h.level;
                //level 肯定是比 oldLevel 大一的，如果小了说明其他线程更新过表了
                if (level <= oldLevel) 
                    break;
                HeadIndex<K,V> newh = h;
                Node<K,V> oldbase = h.node;
                //正常情况下，循环只会执行一次，如果由于其他线程的并发操作导致 oldLevel 的值不稳定，那么会执行多次循环体
                for (int j = oldLevel+1; j <= level; ++j)
                    newh = new HeadIndex<K,V>(oldbase, newh, idxs[j], j);
                //更新头指针
                if (casHead(h, newh)) {
                    h = newh;
                    idx = idxs[level = oldLevel];
                    break;
                }
            }
        }

这一部分的代码主要完成的是根据 level 的值，确认是否需要增加一层索引，如果不需要则构建好底层到 level 层的 index 结点的纵向引用。如果需要，则新创建一层索引，完成 head 结点的指针转移，并构建好纵向的 index 结点引用。

//第三部分
if ((rnd & 0x80000001) == 0){
//省略第二部分的代码段
//第三部分的代码是紧接着第二部分代码段后面的
	splice: for (int insertionLevel = level;;) {
            int j = h.level;
            for (Index<K,V> q = h, r = q.right, t = idx;;) {
	            //其他线程并发操作导致头结点被删除，直接退出外层循环
	            //这种情况发生的概率很小，除非并发量实在太大
                if (q == null || t == null)
                    break splice;
                if (r != null) {
                    Node<K,V> n = r.node;
                    int c = cpr(cmp, key, n.key);
                    //如果 n 正在被其他线程删除，那么调用 unlink 去删除它
                    if (n.value == null) {
                        if (!q.unlink(r))
                            break;
                        //重新获取 q 的右结点，再次进入循环
                        r = q.right;
                        continue;
                    }
                    //c > 0 说明前驱结点定位有误，重新进入
                    if (c > 0) {
                        q = r;
                        r = r.right;
                        continue;
                    }
                }
                if (j == insertionLevel) {
		            //尝试着将 t 插在 q 和 r 之间，如果失败了，退出内循环重试
                    if (!q.link(r, t))
                        break; // restart
                    //如果插入完成后，t 结点被删除了，那么结束插入操作
                    if (t.node.value == null) {
                        findNode(key);
                        break splice;
                    }
                    // insertionLevel-- 处理底层链接
                    if (--insertionLevel == 0)
                        break splice;
                }
				//--j，j 应该与 insertionLevel 同步，它代表着我们创建的那个纵向的结点数组的索引
				//并完成层次下移操作
                if (--j >= insertionLevel && j < level)
                    t = t.down;
                //至此，新节点在当前层次的前后引用关系已经被链接完成，现在处理下一层
                q = q.down;
                r = q.right;
            }
        }
	}
	return null;
}

我们根据概率算法得到了一个 level 值，并且通过第二步创建了 level 个新节点并构成了一个纵向的引用关联，但是这些纵向的结点并没有链接到每层中。而我们的第三部分代码就是完成的这个工作，将我们的新节点在每个索引层都构建好前后的链接关系。下面用三张图描述着三个部分所完成的主要工作。

初始化的跳表如下：

第一部分，新增一个结点到最底层的链表上。

第二部分，假设概率得出一个 level 值为 10，那么根据跳表的算法描述需要新建一层索引层。

第三步，链接各个索引层次上的新节点。

这样就完成了新增结点到跳表中的全部过程，大体上已如上图描述，至于 ConcurrentSkipListMap 中关于并发处理的细节之处，图中无法展示，大家可据此重新感受下源码的实现过程。下面我们着重描述下整个 doPut 方法中还涉及的其他几个方法的具体实现。

首先是 findPredecessor 方法，我们说该方法将根据给定的 key，为我们返回最合适的前驱节点。

private Node<K,V> findPredecessor(Object key, Comparator<? super K> cmp) {
    if (key == null)
        throw new NullPointerException(); 
    for (;;) {
        for (Index<K,V> q = head, r = q.right, d;;) {
            //r 为空说明 head 后面并没有其他节点了
            if (r != null) {
                Node<K,V> n = r.node;
				// r 节点处于待删除状态，那么尝试 unlink 它，失败了将重新进入循环再此尝试
				//否则重新获取 q 的右结点并重新进入循环查找前驱节点
                if (n.value == null) {
                    if (!q.unlink(r))
                        break;           // restart
                    r = q.right;         // reread r
                    continue;
                }
                //大于零说明当前位置上的 q 还不是我们要的前驱节点，继续往后找
                if (cpr(cmp, key, k) > 0) {
                    q = r;
                    r = r.right;
                    continue;
                }
            }
            //如果当前的 level 结束了或者 cpr(cmp, key, k) <= 0 会达到此位置
            //往低层递归，如果没有低层了，那么当前的 q 就是最合适的前驱节点
            //整个循环只有这一个出口，无论如何最终都会从此处结束方法
            if ((d = q.down) == null)
                return q.node;
           //否则向低层递归并重置 q 和 r
            q = d;
            r = d.right;
        }
    }
}

最后总结下 findPredecessor 方法的大体逻辑，首先程序会从 head 节点开始在当前的索引层上寻找最后一个比给定 key 小的结点，它就是我们需要的前驱节点（q），我们只需要返回它即可。

其次我们看看 helpDelete 方法，当检测到某个结点的 value 属性值为 null 的时候，一般都会调用这个方法来删除该结点。

/*
   一般的调用形式如下：
   n.helpDelete(b, f);
*/
void helpDelete(Node<K,V> b, Node<K,V> f) {
    if (f == next && this == b.next) {
       if (f == null || f.value != f) 
            casNext(f, new Node<K,V>(f));
        else
            b.casNext(this, f.next);
    }
}

该方法是 Node 结点的内部实例方法，逻辑相对简单，此处不再赘述。通过该方法可以完成将 b.next 指向 f，完成对 n 结点的删除。

至此，有关 put 方法的源码分析就简单到这，大部分的代码还是用于实现跳表这种数据结构的构建和插入，关于并发的处理，你会发现基本都是双层 for 循环+ CAS 无锁式更新，如果遇到竞争失利将退出里层循环重新进行尝试，否则成功的话就会直接 return 或者退出外层循环并结束 CAS 操作。下面我们看删除操作是如何实现的。

五、remove 并发删除操作的内部实现

remove 方法的部分内容我们在介绍相关预备知识中已经提及过，此处的理解想必会容易些。

public V remove(Object key) {
    return doRemove(key, null);
}

//代码比较多，建议读者结合自己的 jdk 源码共同来分析
final V doRemove(Object key, Object value) {
    if (key == null)
        throw new NullPointerException();
    Comparator<? super K> cmp = comparator;
    outer: for (;;) {
		//找到 key 的前驱节点
		//因为删除不单单是根据 key 找到对应的结点，然后赋 null 就完事的，还要负责链接该结点前后的关联
        for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
            Object v; int c;
            //目前 n 基本上就是我们要删除的结点，它为 null，那自然不用继续了，已经被删除了
            if (n == null)
                break outer;
            Node<K,V> f = n.next;
            //再次确认 n 还是不是 b 的后继结点，如果不是将退出里层循环重新进入
            if (n != b.next)               
                break;
            //如果有人正在删除 n，那么帮助它删除
            if ((v = n.value) == null) {     
                n.helpDelete(b, f);
                break;
            }
            //b 被删除了，重新定位前驱节点
            if (b.value == null || v == n)     
                break;
            //正常情况下，key 应该等于 n.key
            //key 大于 n.key 说明我们要找的结点可能在 n 的后面，往后递归即可
            //key 小于 n.key 说明 key 所代表的结点根本不存在
            if ((c = cpr(cmp, key, n.key)) < 0)
                break outer;
            if (c > 0) {
                b = n;
                n = f;
                continue;
            }
            //如果删除是根据键和值两个参数来删除的话，value 是不为 null 的
            //这种情况下，如果 n 的 value 属性不等于我们传入的 value ，那么是不进行删除的
            if (value != null && !value.equals(v))
                break outer;
			//下面三个步骤才是整个删除操作的核心，大致的逻辑我们也在上文提及过了，此处想必会容易理解些
			//第一步，尝试将待删结点的 value 属性赋值 null，失败将退出重试
            if (!n.casValue(v, null))
                break;
            //第二步和第三步如果有一步由于竞争失败，将调用 findNode 方法根据我们第一步的成果，也就是删除所有 value 为 null 的结点
            if (!n.appendMarker(f) || !b.casNext(n, f))
                findNode(key);  
            //否则说明三个步骤都成功完成了   
            else {
                findPredecessor(key, cmp);  
                //判断此次删除后是否导致某一索引层没有其他节点了，并适情况删除该层索引  
                if (head.right == null)
                    tryReduceLevel();
            }
            @SuppressWarnings("unchecked") V vv = (V)v;
            return vv;
        }
    }
    return null;
}

remove 方法其实从整体上来看，首先会有一堆的判断，根据给定的 key 和 value 会判断是否存在与 key 对应的一个节点，也会判断和待删结点相关的前后结点是否正在被删除，并适情况帮助删除。其次才是删除的三大步骤，核心步骤还是将待删结点的 value 属性赋 null 以标记该结点无用了，至于这个 marker 也是为了分散并发冲突的，最后通过 casNext 完成结点的删除。

六、get 方法获取指定结点的 value

算上本小节将要介绍的 “查” 方法，我们就完成了对并发跳表 “增删改查” 的全部分析。相对于“增”来说，其他的三种操作还是相对容易的，尤其是本小节的“查”操作，下面我们看看它的内部实现：

public V get(Object key) {
    return doGet(key);
}

private V doGet(Object key) {
    if (key == null)
        throw new NullPointerException();
    Comparator<? super K> cmp = comparator;
	//依然是双层循环来处理并发
    outer: for (;;) {
        for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
            Object v; int c;
            //以下的一些判断的作用已经描述了多次，此处不再赘述了
            if (n == null)
                break outer;
            Node<K,V> f = n.next;
            if (n != b.next)           
                break;
            if ((v = n.value) == null) {    
                n.helpDelete(b, f);
                break;
            }
            if (b.value == null || v == n)  
                break;
            //c = 0 说明 n 就是我们要的结点
            if ((c = cpr(cmp, key, n.key)) == 0) {
                @SuppressWarnings("unchecked") V vv = (V)v;
                return vv;
            }
            //c < 0 说明不存在这个 key 所对应的结点
            if (c < 0)
                break outer;
            b = n;
            n = f;
        }
    }
    return null;
}

doGet 方法的实现相对还是比较简单的，所以并没有给出太多的注释，主要还是由于大量的并发判断的代码都是一样的，大多都已经在 doPut 方法中给予了详细的注释了。

七、其它的一些方法的简单描述

//是否包含指定 key 的结点
public boolean containsKey(Object key) {
    return doGet(key) != null;
}

//根据 key 返回该 key 所代表的结点的 value 值，不存在该结点则返回默认的 defaultValue
public V getOrDefault(Object key, V defaultValue) {
    V v;
    return (v = doGet(key)) == null ? defaultValue : v;
}

//返回跳表的实际存储元素个数，采取遍历来进行统计
public int size() {
    long count = 0;
    for (Node<K,V> n = findFirst(); n != null; n = n.next) {
        if (n.getValidValue() != null)
            ++count;
    }
    return (count >= Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int) count;
}

//返回所有键的集
public NavigableSet<K> keySet() {
    KeySet<K> ks = keySet;
    return (ks != null) ? ks : (keySet = new KeySet<K>(this));
}

//返回所有值的集
public Collection<V> values() {
    Values<V> vs = values;
    return (vs != null) ? vs : (values = new Values<V>(this));
}

这里需要说明一点的是，虽然返回来的是键或者值的一个集合，但是无论你是通过这个集合获取键或者值，还是删除集合中的键或者值，都会直接映射到当前跳表实例中。原因是这个集合中没有一个方法是自己实现的，都是调用传入的跳表实例的内部方法，具体的大家查看源码即可知晓，此处不再贴出源码。

至此，有关 SkipList 这种跳表数据结构及其在 jdk 中的实现，以及它的并发版本 ConcurrentSkipListMap 的实现，我们都已经简单的分析完了，有理解错误之处，望指出，相互学习！

参考的几篇优秀博文

Java并发容器之SkipList（需要科学上网）

深入Java集合学习系列：ConcurrentSkipListMap实现原理

Java多线程（四）之ConcurrentSkipListMap深入分析

JavaFx

Posted on 2019-06-06 Edited on 2026-08-23 In Flink

basic

UI框架

https://dlsc.com/products/calendarfx/

字节码增强技术

Posted on 2019-06-06 Edited on 2026-08-23 In Flink

字节码增强技术

ASM
JavaAssist
ByteBuddy
CGLIB

https://juejin.cn/post/6844903970658320391

https://www.wwwbuild.net/E-MapReduce_Spark/22025.html

https://blog.csdn.net/junerli/article/details/112687385

Presto使用了一个基于ASM的字节码生成工具airlift-bytecode（facebook内部库）来进行字节码的构建

Spark 和 Flink 都是使用 Janino 来做codeGen的，
Presto是使用airlift-bytecode

// org.apache.flink.table.runtime.generated.CompileUtils#doCompile
private static <T> Class<T> doCompile(ClassLoader cl, String name, String code) {
		checkNotNull(cl, "Classloader must not be null.");
		CODE_LOG.debug("Compiling: {} \n\n Code:\n{}", name, code);
		SimpleCompiler compiler = new SimpleCompiler();
		compiler.setParentClassLoader(cl);
		try {
			compiler.cook(code);
		} catch (Throwable t) {
			System.out.println(addLineNumber(code));
			throw new InvalidProgramException(
				"Table program cannot be compiled. This is a bug. Please file an issue.", t);
		}
		try {
			//noinspection unchecked
			return (Class<T>) compiler.getClassLoader().loadClass(name);
		} catch (ClassNotFoundException e) {
			throw new RuntimeException("Can not load class " + name, e);
		}
	}

Spring如何实现AOP，请不要再说cglib了！
https://juejin.cn/post/6844903970658320391

Spring AOP原理分析一次看懂 - 知乎
https://zhuanlan.zhihu.com/p/36617574

Spring系列之AOP的原理及手动实现
https://juejin.cn/post/6844903744530808840

【Spring基础】CGLIB动态代理实现原理_yhl_jxy的博客-CSDN博客_cglib动态代理
https://blog.csdn.net/yhl_jxy/article/details/80633194

Spring源码剖析5：JDK和cglib动态代理原理详解 - SegmentFault 思否
https://segmentfault.com/a/1190000020178773

CGLIB(Code Generation Library) 介绍与原理 | 菜鸟教程
https://www.runoob.com/w3cnote/cglibcode-generation-library-intro.html

JDK动态代理生成class文件和cglib动态代理生成class文件_RainSun-CSDN博客_cglib生成的class文件
https://blog.csdn.net/u010811939/article/details/80763336

Java动态生成类以及动态添加属性 - 有梦就能实现 - 博客园
https://www.cnblogs.com/firstdream/p/10078673.html

JaninoCompiler

Posted on 2019-06-06 Edited on 2026-08-23 In Flink

JaninoCompiler

http://janino-compiler.github.io/janino/

Janino is a super-small, super-fast Java™ compiler.

极小、极快、开源，不依赖JDK(只依赖JRE，free)的动态编译器

从内存获取classes文件，不依赖IO

类似的工具:

Jasper: 需要依赖JDK
Eclipse compiler

使用场景

类似Javac，Java源文件文件编译成字节码文件
编译Java表达式、块、类和源码文件
静态代码分析和代码操作
性能优化
1. 动态编译类取代Java代理类，避免所有的反射，显著提高性能。
2. Codegen
规则引擎: 取代DSL(预定义语言: Domain-Specific Language)
Codegen

Janino应用

Kettle: ETL工具 [Kettle-JavaNode]
slf4j + log4j/logback [[JaninoEventEvaluator]]
Apache Commons JCL
JBoss Rules/Drools
Flink SQL
Spark SQL

问题:

能否动态卸载?

使用教程

<dependency>  
    <groupId>org.codehaus.janino</groupId>
    <artifactId>janino</artifactId>
    <version>3.1.6</version>
</dependency>

// 首先定义一个表达式模拟器ExpressionEvaluator对象
ExpressionEvaluator ee = new ExpressionEvaluator();

// 定义一个算术表达式，表达式中需要有2个int类型的参数a和b
String expression = "2 * (a + b)";
ee.setParameters(new String[] { "a", "b" }, new Class[] { int.class, int.class });

// 设置表达式的返回结果也为int类型
ee.setExpressionType(int.class);

// 这里处理（扫描，解析，编译和加载）上面定义的算数表达式.
ee.cook(expression);

// 根据输入的a和b参数执行实际的表达式计算过程
int result = (Integer) ee.evaluate(new Object[] { 19, 23 });
System.out.println(expression + " = " + result);

ScriptEvaluator se = new ScriptEvaluator();
se.setParameters(new String[] { "arg1", "arg2" }, new Class[] { String.class, int.class });
se.cook(
    "System.out.println(arg1);\n"
    + "System.out.println(arg2);\n"
    + "\n"
    + "static void method1() {\n"
    + "    System.out.println(\"run in method1()\");\n"
    + "}\n"
    + "\n"
    + "public static void method2() {\n"
    + "    System.out.println(\"run in method2()\");\n"
    + "}\n"
    + "\n"
    + "method1();\n"
    + "method2();\n"
);
se.evaluate(new Object[]{"aaa",22});

public interface Foo {
    int bar(int a, int b);
}

public static void main(String[] args) throws Exception {
    Foo f = (Foo) ClassBodyEvaluator.createFastClassBodyEvaluator(
        new Scanner(null, new StringReader("public int bar(int a, int b) { return a + b; }")),
        Foo.class,                  // 实现的父类或接口
        (ClassLoader) null          // 这里设置为null表示使用当前线程的class loader
    );
    System.out.println("1 + 2 = " + f.bar(1, 2));

}

深度分析：锁升级过程和锁状态，看完这篇你就懂了！

Posted on 2019-06-06 Edited on 2026-08-23 In Flink

深度分析：锁升级过程和锁状态，看完这篇你就懂了！

转载自 https://segmentfault.com/a/1190000022904663

一、前言

锁的状态总共有四种，级别由低到高依次为：无锁、偏向锁、轻量级锁、重量级锁
其实在 JDK 1.6之前，synchronized 还是一个重量级锁，是一个效率比较低下的锁，
但是在JDK 1.6后，JVM为了提高锁的获取与释放效率对（synchronized ）进行了优化，引入了 偏向锁 和 轻量级锁 ，
从此以后锁的状态就有了四种（无锁、偏向锁、轻量级锁、重量级锁），并且四种状态会随着竞争的情况逐渐升级，而且是不可逆的过程，即不可降级，
也就是说只能进行锁升级（从低级别到高级别），不能锁降级（高级别到低级别），意味着偏向锁升级成轻量级锁后不能降级成偏向锁。
这种锁升级却不能降级的策略，目的是为了提高获得锁和释放锁的效率。

二、锁的四种状态

在 synchronized 最初的实现方式是 “阻塞或唤醒一个Java线程需要操作系统切换CPU状态来完成，这种状态切换需要耗费处理器时间，如果同步代码块中内容过于简单，这种切换的时间可能比用户代码执行的时间还长”，这种方式就是 synchronized实现同步最初的方式，这也是当初开发者诟病的地方，这也是在JDK6以前 synchronized效率低下的原因，JDK6中为了减少获得锁和释放锁带来的性能消耗，引入了“偏向锁”和“轻量级锁”。

所以目前锁状态一种有四种，从级别由低到高依次是：无锁、偏向锁，轻量级锁，重量级锁，锁状态只能升级，不能降级

如图所示：

三、锁状态的思路以及特点

四、锁对比

五、Synchronized锁

synchronized 用的锁是存在Java对象头里的，那么什么是对象头呢？

5.1 Java 对象头

我们以 Hotspot 虚拟机为例，Hopspot 对象头主要包括两部分数据：Mark Word（标记字段）和 Klass Pointer（类型指针）

Mark Word：默认存储对象的HashCode，分代年龄和锁标志位信息。这些信息都是与对象自身定义无关的数据，所以Mark Word被设计成一个非固定的数据结构以便在极小的空间内存存储尽量多的数据。它会根据对象的状态复用自己的存储空间，也就是说在运行期间Mark Word里存储的数据会随着锁标志位的变化而变化。

Klass Point：对象指向它的类元数据的指针，虚拟机通过这个指针来确定这个对象是哪个类的实例。

在上面中我们知道了，synchronized 用的锁是存在Java对象头里的，那么具体是存在对象头哪里呢？答案是：存在锁对象的对象头的Mark Word中，那么MarkWord在对象头中到底长什么样，它到底存储了什么呢？

在64位的虚拟机中：

在32位的虚拟机中：

下面我们以 32位虚拟机为例，来看一下其 Mark Word 的字节具体是如何分配的

无锁 ：对象头开辟 25bit 的空间用来存储对象的 hashcode ，4bit 用于存放对象分代年龄，1bit 用来存放是否偏向锁的标识位，2bit 用来存放锁标识位为01

偏向锁：在偏向锁中划分更细，还是开辟 25bit 的空间，其中23bit 用来存放线程ID，2bit 用来存放 Epoch，4bit 存放对象分代年龄，1bit 存放是否偏向锁标识， 0表示无锁，1表示偏向锁，锁的标识位还是01

轻量级锁：在轻量级锁中直接开辟 30bit 的空间存放指向栈中锁记录的指针，2bit 存放锁的标志位，其标志位为00

重量级锁：在重量级锁中和轻量级锁一样，30bit 的空间用来存放指向重量级锁的指针，2bit 存放锁的标识位，为11

GC标记：开辟30bit 的内存空间却没有占用，2bit 空间存放锁标志位为11。

其中无锁和偏向锁的锁标志位都是01，只是在前面的1bit区分了这是无锁状态还是偏向锁状态

关于内存的分配，我们可以在git中openJDK中 markOop.hpp 可以看出：

public:
  // Constants
  enum { age\_bits                 = 4,
         lock\_bits                = 2,
         biased\_lock\_bits         = 1,
         max\_hash\_bits            = BitsPerWord - age\_bits - lock\_bits - biased\_lock\_bits,
         hash\_bits                = max\_hash\_bits > 31 ? 31 : max\_hash\_bits,
         cms\_bits                 = LP64\_ONLY(1) NOT\_LP64(0),
         epoch\_bits               = 2
  };

age_bits：就是我们说的分代回收的标识，占用4字节
lock_bits：是锁的标志位，占用2个字节
biased_lock_bits：是是否偏向锁的标识，占用1个字节
max_hash_bits：是针对无锁计算的hashcode 占用字节数量，如果是32位虚拟机，就是 32 - 4 - 2 -1 = 25 byte，如果是64 位虚拟机，64 - 4 - 2 - 1 = 57 byte，但是会有 25 字节未使用，所以64位的 hashcode 占用 31 byte
hash_bits：是针对 64 位虚拟机来说，如果最大字节数大于 31，则取31，否则取真实的字节数
cms_bits：不是64位虚拟机就占用 0 byte，是64位就占用 1byte
epoch_bits：就是 epoch 所占用的字节大小，2字节。

5.2 Monitor

Monitor 可以理解为一个同步工具或一种同步机制，通常被描述为一个对象。每一个 Java 对象就有一把看不见的锁，称为内部锁或者 Monitor 锁。

Monitor 是线程私有的数据结构，每一个线程都有一个可用 monitor record 列表，同时还有一个全局的可用列表。每一个被锁住的对象都会和一个 monitor 关联，同时 monitor 中有一个 Owner 字段存放拥有该锁的线程的唯一标识，表示该锁被这个线程占用。

Synchronized是通过对象内部的一个叫做监视器锁（monitor）来实现的，监视器锁本质又是依赖于底层的操作系统的 Mutex Lock（互斥锁）来实现的。而操作系统实现线程之间的切换需要从用户态转换到核心态，这个成本非常高，状态之间的转换需要相对比较长的时间，这就是为什么 Synchronized 效率低的原因。因此，这种依赖于操作系统 Mutex Lock 所实现的锁我们称之为重量级锁。

随着锁的竞争，锁可以从偏向锁升级到轻量级锁，再升级的重量级锁（但是锁的升级是单向的，也就是说只能从低到高升级，不会出现锁的降级）。JDK 1.6中默认是开启偏向锁和轻量级锁的，我们也可以通过-XX:-UseBiasedLocking=false来禁用偏向锁。

六、锁的分类

6.2 无锁

无锁是指没有对资源进行锁定，所有的线程都能访问并修改同一个资源，但同时只有一个线程能修改成功。

无锁的特点是修改操作会在循环内进行，线程会不断的尝试修改共享资源。如果没有冲突就修改成功并退出，否则就会继续循环尝试。如果有多个线程修改同一个值，必定会有一个线程能修改成功，而其他修改失败的线程会不断重试直到修改成功。

6.3 偏向锁

初次执行到synchronized代码块的时候，锁对象变成偏向锁（通过CAS修改对象头里的锁标志位），字面意思是“偏向于第一个获得它的线程”的锁。执行完同步代码块后，线程并不会主动释放偏向锁。当第二次到达同步代码块时，线程会判断此时持有锁的线程是否就是自己（持有锁的线程ID也在对象头里），如果是则正常往下执行。由于之前没有释放锁，这里也就不需要重新加锁。如果自始至终使用锁的线程只有一个，很明显偏向锁几乎没有额外开销，性能极高。

偏向锁是指当一段同步代码一直被同一个线程所访问时，即不存在多个线程的竞争时，那么该线程在后续访问时便会自动获得锁，从而降低获取锁带来的消耗，即提高性能。

当一个线程访问同步代码块并获取锁时，会在 Mark Word 里存储锁偏向的线程 ID。在线程进入和退出同步块时不再通过 CAS 操作来加锁和解锁，而是检测 Mark Word 里是否存储着指向当前线程的偏向锁。轻量级锁的获取及释放依赖多次 CAS 原子指令，而偏向锁只需要在置换 ThreadID 的时候依赖一次 CAS 原子指令即可。

偏向锁只有遇到其他线程尝试竞争偏向锁时，持有偏向锁的线程才会释放锁，线程是不会主动释放偏向锁的。

关于偏向锁的撤销，需要等待全局安全点，即在某个时间点上没有字节码正在执行时，它会先暂停拥有偏向锁的线程，然后判断锁对象是否处于被锁定状态。如果线程不处于活动状态，则将对象头设置成无锁状态，并撤销偏向锁，恢复到无锁（标志位为01）或轻量级锁（标志位为00）的状态。

6.4 轻量级锁（自旋锁）

轻量级锁是指当锁是偏向锁的时候，却被另外的线程所访问，此时偏向锁就会升级为轻量级锁，
其他线程会通过自旋（关于自旋的介绍见文末）的形式尝试获取锁，
线程不会阻塞，从而提高性能。

轻量级锁的获取主要由两种情况：

① 当关闭偏向锁功能时；
② 由于多个线程竞争偏向锁导致偏向锁升级为轻量级锁。

一旦有第二个线程加入锁竞争，偏向锁就升级为轻量级锁（自旋锁）。

这里要明确一下什么是锁竞争：
如果多个线程轮流获取一个锁，但是每次获取锁的时候都很顺利，没有发生阻塞，那么就不存在锁竞争。
只有当某线程尝试获取锁的时候，发现该锁已经被占用，只能等待其释放，这才发生了锁竞争。

在轻量级锁状态下继续锁竞争，没有抢到锁的线程将自旋，即不停地循环判断锁是否能够被成功获取。
获取锁的操作，其实就是通过CAS修改对象头里的锁标志位。先比较当前锁标志位是否为“释放”，
如果是则将其设置为“锁定”，比较并设置是原子性发生的。这就算抢到锁了，然后线程将当前锁的持有者信息修改为自己。

长时间的自旋操作是非常消耗资源的，一个线程持有锁，其他线程就只能在原地空耗CPU，执行不了任何有效的任务，这种现象叫做忙等（busy-waiting）。
如果多个线程用一个锁，但是没有发生锁竞争，或者发生了很轻微的锁竞争，那么synchronized就用轻量级锁，允许短时间的忙等现象。
这是一种折衷的想法，短时间的忙等，换取线程在用户态和内核态之间切换的开销。

6.4 重量级锁

重量级锁显然，此忙等是有限度的（有个计数器记录自旋次数，默认允许循环10次，可以通过虚拟机参数更改）。
如果锁竞争情况严重，某个达到最大自旋次数的线程，会将轻量级锁升级为重量级锁（依然是CAS修改锁标志位，但不修改持有锁的线程ID）。
当后续线程尝试获取锁时，发现被占用的锁是重量级锁，则直接将自己挂起（而不是忙等），等待将来被唤醒。

重量级锁是指当有一个线程获取锁之后，其余所有等待获取该锁的线程都会处于阻塞状态。

简言之，就是所有的控制权都交给了操作系统，由操作系统来负责线程间的调度和线程的状态变更。而这样会出现频繁地对线程运行状态的切换，
线程的挂起和唤醒，从而消耗大量的系统资