发表于

本文已参与好文召集令活动，点击查看：后端、大前端双赛道投稿，2万元奖池等你挑战！

首先分享之前的所有文章 , 欢迎点赞收藏转发三连下次一定 >>>> 😜😜😜

文章合集 : 🎁 juejin.cn/post/694164…

Github : 👉 github.com/black-ant

CASE 国内备份 : 👉 gitee.com/antblack/ca…

一 .前言

Java 流这个特性已经出来很久了 , 可以大大的减少我们的代码 , 而且并行处理可以在某些场景下使用多个处理器核心可以大大的提高性能.

不过 Stream 语法新手使用起来还是会有一定的难度 , 这一篇文档由浅到深看一下这个特性.

这是开篇 , 只记录之前梳理的用法 , 下一篇来看源码 , 记得收藏!!!!!

Stream 的特点

  • Stream 不是集合 , 也不是数据结构 , 不可以保存数据
  • Stream 有点类似于高级 的 Iterator , 可以用于算法和计算
  • 不同于迭代器 , Stream 可以并行化操作 , 数据被分为很多段 , 在不同的线程中进行处理
  • 数据源、零个或多个中间操作 ( intermediate ) 以及零个或一个终端操作 (terminal )
  • 所有中间操作都是惰性的 , 在管道开始工作之前，任何操作都不会产生任何效果
  • 终端操作有点像水龙头 , 开启了水龙头后 , 水才会流动 , 中间操作才会执行

image.png

二. 基础知识

2.1 结构运算

2.1.1 双冒号运算

双冒号运算就是将方法当成参数传递给需要的方法 ( Stream ) , 即为方法引用

案例一 : 基础用法

1
2
3
java复制代码x -> System.out.println(x)
// ------------ 
System.out::println

案例二 : 复杂用法

1
2
3
4
5
java复制代码for (String item: list) {
    AcceptMethod.printValur(item);
}
//------------------
list.faorEach(AcceptMethod::printValur);

2.2 流的创建

2.2.1 集合和数组工具

基础案例

1
2
3
4
5
6
7
8
9
10
java复制代码// Collection 工具
Collection.stream () : list.stream();

Stream.<String>builder().add("a").add("b").add("c").build();

Stream.of("a", "b", "c")

Stream.generate(() -> "element").limit(10);

Stream.iterate(40, n -> n + 2).limit(20);

创建一个整数流

1
2
3
4
5
java复制代码IntStream.rangeClosed(1, 100).reduce(0, Integer::sum);
IntStream.rangeClosed(1, 100).parallel().reduce(0, Integer::sum);

// 其他的基本类型案例
LongStream.rangeClosed(1, 3);

创建一个并行流

1
2
3
4
5
6
7
8
9
10
java复制代码// API :
Stream<E> parallelStream()

// 案例 :
Collection.parallelStream ()
listOfNumbers.parallelStream().reduce(5, Integer::sum);

listOfNumbers.parallelStream().forEach(number ->
    System.out.println(number + " " + Thread.currentThread().getName())
);

数组创建流

1
2
3
4
java复制代码Arrays.stream(intArray).reduce(0, Integer::sum);
Arrays.stream(intArray).parallel().reduce(0, Integer::sum);
Arrays.stream(integerArray).reduce(0, Integer::sum);
Arrays.stream(integerArray).parallel().reduce(0, Integer::sum);

合并流

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
JAVA复制代码// API : 组合2个 Streams
 <T> Stream<T> concat(Stream<? extends T> a, Stream<? extends T> b)
     
     
// 案例
Stream<Integer> stream1 = Stream.of(1, 3, 5);
Stream<Integer> stream2 = Stream.of(2, 4, 6);

Stream<Integer> resultingStream = Stream.concat(stream1, stream2);


// 案例 ：　合并三个
Stream.concat(Stream.concat(stream1, stream2), stream3);

// 案例 : stream of 合并流
Stream.of(stream1, stream2, stream3, stream4)

其他的案例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
java复制代码// 静态工厂
1.  Java.util.stream.IntStream.range ( )
2.  Java.nio.file.Files.walk ( )

// 手动创建
1.  java.util.Spliterator
2.  Random.ints()
3.  BitSet.stream()
4.  Pattern.splitAsStream(java.lang.CharSequence)
5.  JarFile.stream()
    
// java.io.BufferedReader.lines()
Files.lines(path, Charset.forName("UTF-8"));
Files.lines(path);

补充

分割流案例

image.png

2.3 流的操作

一个流可以有多个 intermediate 操作 , 和一个 Terminal 操作 , 当 Terminal 执行完成后, 流就结束了

2.3.1 流的 Intermediate 操作

map : 元素映射

1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码// API :
<R> Stream<R> map(Function<? super T, ? extends R> mapper)
    


// Map 传入方法函数 , Map 返回的是一个 object
books.stream().filter(e -> "Effective Java".equals(e.getValue())).map(Map.Entry::getKey).findFirst();

wordList.stream().map(String::toUpperCase).collect(Collectors.toList());

Stream.of(1, 2, 3).map(n -> n + 1).collect(Collectors.toList());

nums.stream().map( n -> n * n ).collect (Collectors.toList());

flatMap

1
2
3
4
5
6
7
8
9
10
java复制代码// flatMap 返回的是一个 Stream
Stream<List<String>> namesOriginalList = Stream.of(
	Arrays.asList("Pankaj"), 
	Arrays.asList("David", "Lisa"),
	Arrays.asList("Amit"));
//flat the stream from List<String> to String stream
Stream<String> flatStream = namesOriginalList
	.flatMap(strList -> strList.stream());

flatStream.forEach(System.out::println);

mapToXXX

1
2
3
4
5
6
7
8
9
10
java复制代码// API : 
IntStream mapToInt(ToIntFunction<? super T> mapper)

// 作用 : mapToXXX 主要用于转换为

doubleNumbers.stream().mapToDouble(Double::doubleValue).sum();

customers.stream().mapToInt(Customer::getAge).filter(c -> c > 65).count();

intStream1.mapToObj(c -> (char) c);

filter : 过滤 ,通过过滤的元素被流下来生成新的 stream

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
java复制代码// Predicate 是一个函数式接口
Stream<T> filter(Predicate<? super T> predicate)

// filter 中使用箭头表达式
- tream<Integer> oddIntegers = ints.stream().filter(i -> i.intValue() % 2 != 0);
- list.stream().filter(p -> p.startsWith("j")).count()

// Filter 中使用 :: 双冒号
customers.stream().filter(Customer::hasOverHundredPoints).collect(Collectors.toList());

// Filter 中使用代码块
 customers.stream().filter(c -> {
      try {
          return c.hasValidProfilePhoto();
      } catch (IOException e) {
          //handle exception
      }
      return false;
}).collect(Collectors.toList());

distinct : 去重

1
2
java复制代码nums.stream().filter(num -> num % 2 == 0).distinct().collect(Collectors.toList());
list.stream().distinct().collect(Collectors.toList())

sorted : 排序

1
2
3
4
5
6
7
8
java复制代码// 自定义排序方式
persons.stream().limit(2).sorted((p1, p2) -> p1.getName().compareTo(p2.getName())).collect(Collectors.toList());

// 使用指定 Comparator 提供的排序器
List<String> reverseSorted = names2.sorted(Comparator.reverseOrder()).collect(Collectors.toList());

// 不传入参数使用默认排序方式
List<String> naturalSorted = names3.sorted().collect(Collectors.toList());

peek

1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码// API : 可以用于调试 ,主要在流通过管道中的某个点时进行拦截
Stream<T> peek(Consumer<? super T> action)
    
// 案例 :     
IntStream.range(1, 10).peek(System.out::println).sum();

// 在多个拦截点拦截
Stream.of("one", "two", "three", "four")
	.filter(e -> e.length() > 3)
	.peek(e -> System.out.println("Filtered value: " + e))
	.map(String::toUpperCase)
	.peek(e -> System.out.println("Mapped value: " + e))
	.collect(Collectors.toList());

limit : 限制

1
2
3
4
5
java复制代码// API : 截断流的数量 , 可以看到还是返回一个流
Stream<T> limit(long maxSize);

// 案例 : 
nums.stream().filter(n-> n>2).limit(2).collect(Collectors.toList ())

skip : 跳过

1
2
3
4
5
java复制代码// API : 
Stream<T> skip(long n);

// 案例 : 
nums. stream() .filter(n-> n>2 ).skip (2) . collect( Collectors . toList () );

parallel 并行流

1
2
3
4
5
6
java复制代码// API : 返回一个并行的等效流 , 如果已经是并行 ,返回自身
S parallel()
boolean isParallel()

// 案例 : 
Object[] listOutput = list.stream().parallel().toArray();

sequential : 串行流

1
2
3
4
5
java复制代码// API :
S sequential();

// 案例 : 
Arrays.asList(1, 2, 3, 4).stream().sequential();

unordered : 无序化

1
2
java复制代码// 消除相遇顺序 , 可以提交并行性能
IntStream.range(1, 1_000_000).unordered().parallel().distinct().toArray();

2.3.2 流的 Terminal 操作之聚合

foreach : 循环遍历

1
2
3
4
5
java复制代码// API : 可以看到 , 这里接受到的是一个 Consumer 函数
void forEach(Consumer<? super T> action);

// foreach 中使用箭头函数
roster.stream().forEach(p -> System.out.println(p.getName()));

forEachOrdered : 有序的循环流

1
java复制代码list.stream().parallel().forEachOrdered(e -> logger.info(e));

Array

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
java复制代码stream.toArray(String[]::new);

//reduce : 把 Stream 元素组合起来
Stream.of("A", "B", "C", "D").reduce("", String::concat);

// reduce 求和
Stream.of(5, 6, 7, 8).reduce(0, (accumulator, element) -> accumulator + element);
    ?--- reduce 后面的参数 : 第一个默认值 , 后面是传入的方法


// min : 将字符串数组求最大值
Stream.of(testStrings).max((p1, p2) -> Integer.compare(p1.length(), p2.length()));

// max : 获得最大长度
br.lines().mapToInt(String::length).max().getAsInt();

collection

  • stream.collect(Collectors.toList()) : toList把流中所有的元素收集到List 中
  • stream.collect(Collectors.toCollection(ArrayList::new)): 把流中的元素收集到给定的供应源创建的集合中
  • stream.collect(Collectors.toSet()) : 把流中所有的元素保存到Set集合中,删除重复锁
  • stream.collect(Collectors.toCollection(Stack::new))
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
java复制代码// API
<R> R collect(Supplier<R> supplier,BiConsumer<R, ? super T> accumulator,BiConsumer<R, R> combiner);
<R, A> R collect(Collector<? super T, A, R> collector);




Map<String, Integer> hashMap = list.stream().collect(Collectors
    .toMap(Function.identity(), String::length));

Map<String, Integer> linkedHashMap = list.stream().collect(Collectors.toMap(
    Function.identity(),
    String::length,
    (u, v) -> u,
    LinkedHashMap::new
));

// 创建 Collection 对象
Stream<Integer> intStream = Stream.of(1,2,3,4);
List<Integer> intList = intStream.collect(Collectors.toList());
System.out.println(intList); //prints [1, 2, 3, 4]

intStream = Stream.of(1,2,3,4); //stream is closed, so we need to create it again
Map<Integer,Integer> intMap = intStream.collect(Collectors.toMap(i -> i, i -> i+10));
System.out.println(intMap); //prints {1=11, 2=12, 3=13, 4=14}


// 创建 Array 对象
Stream<Integer> intStream = Stream.of(1,2,3,4);
Integer[] intArray = intStream.toArray(Integer[]::new);
System.out.println(Arrays.toString(intArray)); //prints [1, 2, 3, 4]

// String 操作
stream.collect(Collectors.joining()).toString();
list.stream().collect(Collectors.joining(" | ")) : 连接符中间穿插
list.stream().collect(Collectors.joining(" || ", "Start--", "--End")) : 连接符中间及前后
    
// 创建为 Map
books.stream().collect(Collectors.toMap(Book::getIsbn, Book::getName));
// ps : Collector<T, ?, Map<K,U>> toMap(Function<? super T, ? extends K> keyMapper,Function<? super T, ? extends U> valueMapper)

2.3.3 流的 Terminal 操作之运算

min : 返回此流的最小元素

1
2
3
4
5
6
JAVA复制代码// API
Optional<T> min(Comparator<? super T> comparator) 
Collector<T, ?, Optional<T>> minBy(Comparator<? super T> comparator)

// 案例 :
list.stream().min(Comparator.comparing(String::valueOf)).ifPresent(e -> System.out.println("Min: " + e));

max : 返回此流的最大元素

1
2
3
4
5
6
JAVA复制代码// API 
Optional<T> max(Comparator<? super T> comparator);
Collector<T, ?, Optional<T>> maxBy(Comparator<? super T> comparator) 
    
// 案例 :
list.stream().max(Comparator.comparing(String::valueOf)).ifPresent(e -> System.out.println("Max: " + e));

count : 计算流中的项目数

1
2
3
4
5
JAVA复制代码// API : 
<T> Collector<T, ?, Long>javacounting()

// 案例 :
Stream.of(1,2,3,4,5).count();

reduce

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
java复制代码
// API : reduce 用于 Stream 中进行计算处理
Optional<T> reduce(BinaryOperator<T> accumulator);
Collector<T, ?, T> reducing(T identity, BinaryOperator<T> op) 
U reduce(U identity,BiFunction<U, ? super T, U> accumulator,BinaryOperator<U> combiner);

Collector<T, ?, U> reducing(U identity,Function<? super T, ? extends U> mapper,BinaryOperator<U> op) 

// 参数含义 : 
identity : 缩减的标识值(也是没有输入元素时返回的值)
accumulator : 执行的操作
    


// 使用案例
numbers.reduce((i,j) -> {return i*j;});
numbers.stream().reduce(0, (subtotal, element) -> subtotal + element);
numbers.stream().reduce(0, Integer::sum);

// 关联字符串
letters.stream().reduce("", (partialString, element) -> partialString + element);

// 关联大小写
letters.stream().reduce("", (partialString, element) -> partialString.toUpperCase() + element.toUpperCase());

ages.parallelStream().reduce(0, a, b -> a + b, Integer::sum);


// 并行操作要点 : 并行处理运算必须要符合如下操作
1. 结果不受操作数顺序的影响
2. 互不干扰: 操作不影响数据源
3. 无状态和确定性: 操作没有状态，并且为给定的输入生成相同的输出
userList.parallelStream().reduce(0, (partialAgeResult, user) -> partialAgeResult + user.getAge(), Integer::sum);

2.4 流的 Terminal 操作之搜索

anyMatch：任一元素满足匹配条件

1
2
3
4
5
6
java复制代码// API  : Stream 中只要有一个元素符合传入的 predicate，返回 true
 boolean anyMatch(Predicate<? super T> predicate);


// 案例 : 
persons.stream(). anyMatch(p -> p.getAge() < 12);

allMatch：所有元素都满足匹配条件

1
2
3
4
5
java复制代码// API :  Stream 中全部元素符合传入的 predicate，返回 true
boolean allMatch(Predicate<? super T> predicate);

// 案例 : 
persons.stream(). allMatch(p -> p.getAge() > 18);

findFirst：返回Stream中的第一个元素

1
2
3
4
5
java复制代码// API : 返回一个 Optional 标识第一个元素
Optional<T> findFirst();

// 案例 : 
students.stream().filter(student ->student .getage()>20 ).findFirst();

findAny：返回Stream中的任意个元素

1
2
java复制代码// API : findAny不一定返回第一个，而是返回任意一个 , 如果流为空则返回一个空的Optional
Optional<T> findAny();

noneMatch：所有元素都不满足匹配条件

1
2
3
4
5
java复制代码// API : 当Stream 中没有一个元素符合传入的 predicate，返回 true 
boolean noneMatch(Predicate<? super T> predicate);

// 案例 :
numbers5.noneMatch(i -> i==10)

2.5 流的规约

1
2
3
4
5
6
7
java复制代码// reduce : 对参数化操作后的集合进行进一步操作
    
students.stream().filter(student -> "计算机科学".equals(student.getMajor())).map(Student::getAge).reduce(0, (a, b) -> a + b);

students.stream().filter(student -> "计算机科学".equals(student.getMajor())).map(Student::getAge).reduce(0, Integer::sum);

students.stream().filter(student -> "计算机科学".equals(student.getMajor())).map(Student::getAge).reduce(Integer::sum);

2.6 流的分组 (Group By)

单级分组

1
2
3
4
java复制代码// API : 
public static <T, K> Collector<T, ?, Map<K, List<T>>>groupingBy(Function<? super T, ? extends K> classifier)

students.stream().collect(Collectors.groupingBy(Student::getSchool))

多级分组

1
2
3
4
5
6
java复制代码// 作用 :  

// 案例 : 
students.stream().collect(
    Collectors.groupingBy(Student::getSchool,
    Collectors.groupingBy(Student::getMajor)));

partitioningBy : 分区 ,区别于groupBy ,分区中只有true , false .

1
2
3
4
5
6
java复制代码// API : 可以看到 , 这里主要是 Predicate 函数
Collector<T, ?, Map<Boolean, List<T>>> partitioningBy(Predicate<? super T> predicate)
Collector<T, ?, Map<Boolean, D>> partitioningBy(Predicate<? super T> predicate,Collector<? super T, A, D> downstream)
    
 // 案例 :    
students.stream().collect(Collectors.partitioningBy(student -> "武汉大学".equals(student.getSchool())));

三 . 使用深入

3.1 扩展线程池

1
2
3
4
5
6
7
8
java复制代码// 全局设置线程池
-D java.util.concurrent.ForkJoinPool.common.parallelism=4

// 手动设置线程此
ForkJoinPool customThreadPool = new ForkJoinPool(4);
int sum = customThreadPool.submit(
    () -> listOfNumbers.parallelStream().reduce(0, Integer::sum)).get();
customThreadPool.shutdown();

3.2 Debug 处理

四 . 常见案例

案例 一 : 对 List AList 中 每个 元素进行操作转换后生成 另外的类型 C 放入 List BList

1
2
3
4
5
6
7
java复制代码List<JSONObject> itemjson = new LinkedList<JSONObject>();
List<A> aList = ... 
itemCW.stream().map(c -> {
            JSONObject nodeitem = new JSONObject();
            nodeitem.put("whtype", 0);
            return nodeitem;
}).forEach(c -> itemjson.add(c));

案例二 ： 对 Map 进行循环操作

1
2
3
4
5
java复制代码       realmTO.getTemplates().forEach((key, template) -> {
            AnyType type = anyTypeDAO.find(key);
            anyTemplate.set(template);
        });
// for-each  的时候 ， 可以传入 key 和 对象 ，后续可以使用

案例三 ： 从大集合中获取小集合

1
2
3
4
5
6
7
java复制代码
// 获取id的集合
List<Long> idList = stockList.stream().map(Stock::getId).collect(Collectors.toList());
// 获取skuid集合并去重
List<Long> skuIdList = stockList.stream().map(Stock::getSkuId).distinct().collect(Collectors.toList());
// 获取supplierId集合(supplierId的类型为int，返回List<Integer>,使用boxed方法装箱)
Set<Integer> supplierIdSet = stockList.stream().mapToInt(Stock::getSupplierId).boxed().collect(Collectors.toSet());

案例四 ： 分组与分片

1
2
3
4
5
6
java复制代码// 按skuid分组
Map<Long, List<Stock>> skuIdStockMap = stockList.stream().collect(Collectors.groupingBy(Stock::getSkuId));
// 过滤supplierId=1然后按skuId分组
Map<Long, List<Stock>> filterSkuIdStockMap = stockList.stream().filter(s -> s.getSupplierId() == 1).collect(Collectors.groupingBy(Stock::getSkuId));
// 按状态分为不可用和其他两个分片
Map<Boolean, List<Stock>> partitionStockMap = stockList.stream().collect(Collectors.partitioningBy(s -> s.getStatus() == 0));

案例五 ： 计数与求和

1
2
3
4
java复制代码// 统计skuId=1的记录数
long skuIdRecordNum = stockList.stream().filter(s -> s.getSkuId() == 1).count();
// 统计skuId=1的总库存量
BigDecimal skuIdAmountSum = stockList.stream().filter(s -> s.getSkuId() == 1).map(Stock::getAmount).reduce(BigDecimal.ZERO, BigDecimal::add);

案例 六 ：特定用法

1
2
3
4
5
6
7
8
java复制代码// 多重分组并排序，先按supplierId分组，再按skuId分组，排序规则，先supplierId后skuId
Map<Integer, Map<Long, List<Stock>>> supplierSkuStockMap = stockList.stream().collect(Collectors.groupingBy(Stock::getSupplierId, TreeMap::new,
                Collectors.groupingBy(Stock::getSkuId, TreeMap::new, Collectors.toList())));

// 多条件排序，先按supplierId正序排，再按skuId倒序排
// (非stream方法，而是集合的sort方法，直接改变原集合元素，使用Function参数)
stockList.sort(Comparator.comparing(Stock::getSupplierId)
                .thenComparing(Stock::getSkuId, Comparator.reverseOrder()));

案例 七 ： 对流进行排序

1
2
3
4
5
6
7
8
9
java复制代码Collections.sort(literals, (final String t, final String t1) -> {
            if (t == null && t1 == null) {
                return 0;
            } else if (t != null && t1 == null) {
                return -1;
            } 
   });

// t1 t2 是其中进行比较的了 2个对象 ，在里面定义相关的排序方法，通过返回 true / false 返回排序规则

案例 八 ： 对流进行 filter 后 ，获取第一个

1
2
3
java复制代码correlationRules.stream().filter(rule -> anyType != null && anyType.equals(rule.getAnyType())).findFirst()
    
 // 关键在于 filter 和 findFirst

案例 九 ：一种类型集合转换为另外一种类型集合

1
2
3
4
5
6
java复制代码        List<String> strings = Lists.transform(list, new Function<Integer, String>() {
            @Override
            public String apply(@Nullable Integer integer) {
                return integer.toString();
            }
        });

案例 十 ： 遍历集合并且返回集合

1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码return Stream.of(resources).map(resource -> preserveSubpackageName(baseUrlString, resource, path)).collect(Collectors.toList());


    private String preserveSubpackageName(final String baseUrlString, final Resource resource, final String rootPath) {
        try {
            return rootPath + (rootPath.endsWith("/") ? "" : "/")
                + resource.getURL().toString().substring(baseUrlString.length());
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

// 注意 ，其中调用了下面的方法 ，直接的匿名方法暂时不会写

案例十一 ：简单拼接

1
2
3
4
5
6
7
8
9
10
11
java复制代码// 拼接成 [x, y, z] 形式
String result1 = stream1.collect(Collectors.joining(", ", "[", "]"));
// 拼接成 x | y | z 形式
String result2 = stream2.collect(Collectors.joining(" | ", "", ""));
// 拼接成 x -> y -> z] 形式
String result3 = stream3.collect(Collectors.joining(" -> ", "", ""));


(String)value.stream().map((i) -> {
                return this.formatSql("{0}", i);
            }).collect(Collectors.joining(",", "(", ")"));

案例十二 : 复杂使用

1
2
3
4
5
6
7
8
java复制代码buzChanlList.stream()
.map(item -> {
      return null;
	})
.filter(item -> {
       return isok;
     })
.forEach(c -> contentsList.add(c));

案例十三 : 切分集合

1
java复制代码 List<List<Integer>> splitList = Stream.iterate(0, n -> n + 1).limit(limit).parallel().map(a -> list.stream().skip(a * MAX_NUMBER).limit(MAX_NUMBER).parallel().collect(Collectors.toList())).collect(Collectors.toList());

案例十四 : filter 操作

1
java复制代码collection.stream().filter(person -> "1".equals(person.getGender())).collect(Collectors.toList())

总结

流的并行特性在使用得当的情况下可以大大增加效率 , 下一篇我们来看一下源码.

参考

@ www.it610.com/article/129…

@ www.baeldung.com/intellij-de…

@ Java 8 Stream - Java Stream - JournalDev

@ www.baeldung.com/java-stream…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

本文已参与好文召集令活动，点击查看：后端、大前端双赛道投稿，2万元奖池等你挑战！」

前言

  版本控制最主要的功能就是追踪文件的变更。它将什么时候、什么人更改了文件的什么内容等信息忠实地了记录下来。本章初识SVN

1、SVN简介

  Subversion(SVN) 是一个开源的版本控制系統, 也就是说 Subversion 管理着随时间改变的数据。 这些数据放置在一个中央资料档案库(repository) 中。 这个档案库很像一个普通的文件服务器, 不过它会记住每一次文件的变动。 这样你就可以把档案恢复到旧的版本, 或是浏览文件的变动历史。

2、SVN 的一些概念

  • repository（源代码库）:源代码统一存放的地方。
  • Checkout（提取）:当没有源代码的时候，需要从repository checkout一份，进行修改和开发。
  • Commit（提交）:当已经修改了代码，需要Commit到repository中。
  • Update (更新):当已经Checkout了一份源代码， Update一下就可以和Repository上的源代码同步。

3、SVN 的主要功能

  • 目录版本控制
  • 真实的版本历史
  • 自动提交
  • 纳入版本控管的元数据
  • 选择不同的网络层
  • 一致的数据处理方式
  • 有效的分支(branch)与标签(tag)
  • Hackability

4、SVN 检出操作

1
js复制代码svn checkout http://svn.test.com/svn/project --username=username

  检出成功后在当前目录下生成检出目录。查看检出的内容 执行
ll project /

5、SVN 解决冲突

  当发现文件存在错误，需要修改文件并提交到版本库中。需要进行解决冲突的操作。
  首先查看文件不同之处

1
js复制代码svn diff

  然后尝试使用下面的命令来提交更改：

1
js复制代码svn commit -m "变更备注"

  提示提交失败，为了避免两人的代码被互相覆盖，Subversion 不允许我们进行这样的操作。对文件进行更新。

1
js复制代码svn update

  此时本地和仓库已经同步，可以安全地提交更改了

1
js复制代码svn commit -m "再次变更备注"

6、SVN 提交操作

  首先将文件放入版本控制中

1
csharp复制代码svn add "需要提交的文件"

  可以查看文件在版本控制中的状态

1
js复制代码svn status

  最后进行提交

1
js复制代码svn commit -m "需要提交的文件"

7、SVN 版本回退

  发现修改错误，要撤销修改，通过 svn revert 文件 回归到未修改状态

1
js复制代码svn revert "需要回退的文件"

  恢复目录

1
js复制代码svn revert -R "目录文件"

  撤销回之前的版本，现在是版本 1008，撤回到1006 版本

1
js复制代码svn merge -r 1008:1006 readme

8、SVN分支管理

  创建分支

1
js复制代码svn copy 版本库/ branches/新建分支

  提交新增的分支到版本库。

1
js复制代码svn commit -m "add 新建分支"

  分支合并

1
js复制代码 svn merge ../branches/新建分支/

9、SVN 查看历史信息

  • svn log: 用来展示svn 的版本作者、日期、路径等等。
  • svn diff: 用来显示特定修改的行级详细信息。
  • svn cat: 取得在特定版本的某文件显示在当前屏幕。
  • svn list: 显示一个目录或某一版本存在的文件。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

本文正在参加「Python主题月」，详情查看活动链接。

有时候，我们需要用到代理IP，比如在爬虫的时候，但是得到了IP之后，可能不知道怎么验证这些IP是不是有效的，这时候我们可以使用Python携带该IP来模拟访问某一个网站，如果多次未成功访问，则说明这个代理是无效的。
代码如下：

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
python复制代码import requests
import random
import time

http_ip = [
    '118.163.13.200:8080',
    '222.223.182.66:8000',
    '51.158.186.242:8811',
    '171.37.79.129:9797',
    '139.255.123.194:4550'
]

for i in range(10):
    try:
        ip_proxy = random.choice(http_ip)
        proxy_ip = {
            'http': ip_proxy,
            'https': ip_proxy,
        }
        print('使用代理的IP:', proxy_ip)
        response = requests.get("http://httpbin.org/ip", proxies=proxy_ip).text
        print(response)
        print('当前IP有效')
        time.sleep(2)
    except Exception as e:
        print(e.args[0])
        print('当前IP无效')
        continue

运行结果如下：

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
python复制代码使用代理的IP: {'http': '118.163.13.200:8080', 'https': '118.163.13.200:8080'}
HTTPConnectionPool(host='118.163.13.200', port=8080): Max retries exceeded with url: http://httpbin.org/ip (Caused by ProxyError('Cannot connect to proxy.', NewConnectionError('<urllib3.connection.HTTPConnection object at 0x00000247674F5F88>: Failed to establish a new connection: [WinError 10060] 由于连接方在一段时间后没有正确答复或连接的主机没有反应，连接尝试失败。')))
当前IP无效
使用代理的IP: {'http': '51.158.186.242:8811', 'https': '51.158.186.242:8811'}
{
  "origin": "51.158.186.242"
}

当前IP有效
使用代理的IP: {'http': '222.223.182.66:8000', 'https': '222.223.182.66:8000'}
{
  "origin": "139.202.62.84, 222.223.182.66"
}

当前IP有效
使用代理的IP: {'http': '51.158.186.242:8811', 'https': '51.158.186.242:8811'}
{
  "origin": "51.158.186.242"
}

当前IP有效
使用代理的IP: {'http': '51.158.186.242:8811', 'https': '51.158.186.242:8811'}
{
  "origin": "51.158.186.242"
}

当前IP有效
使用代理的IP: {'http': '222.223.182.66:8000', 'https': '222.223.182.66:8000'}
HTTPConnectionPool(host='222.223.182.66', port=8000): Max retries exceeded with url: http://httpbin.org/ip (Caused by ProxyError('Cannot connect to proxy.', NewConnectionError('<urllib3.connection.HTTPConnection object at 0x00000247675067C8>: Failed to establish a new connection: [WinError 10060] 由于连接方在一段时间后没有正确答复或连接的主机没有反应，连接尝试失败。')))
当前IP无效
使用代理的IP: {'http': '139.255.123.194:4550', 'https': '139.255.123.194:4550'}
HTTPConnectionPool(host='139.255.123.194', port=4550): Max retries exceeded with url: http://httpbin.org/ip (Caused by ProxyError('Cannot connect to proxy.', NewConnectionError('<urllib3.connection.HTTPConnection object at 0x00000247674F55C8>: Failed to establish a new connection: [WinError 10060] 由于连接方在一段时间后没有正确答复或连接的主机没有反应，连接尝试失败。')))
当前IP无效
使用代理的IP: {'http': '51.158.186.242:8811', 'https': '51.158.186.242:8811'}
{
  "origin": "51.158.186.242"
}

当前IP有效
使用代理的IP: {'http': '51.158.186.242:8811', 'https': '51.158.186.242:8811'}
{
  "origin": "51.158.186.242"
}

当前IP有效
使用代理的IP: {'http': '222.223.182.66:8000', 'https': '222.223.182.66:8000'}
HTTPConnectionPool(host='222.223.182.66', port=8000): Max retries exceeded with url: http://httpbin.org/ip (Caused by ProxyError('Cannot connect to proxy.', NewConnectionError('<urllib3.connection.HTTPConnection object at 0x0000024767514908>: Failed to establish a new connection: [WinError 10060] 由于连接方在一段时间后没有正确答复或连接的主机没有反应，连接尝试失败。')))
当前IP无效

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

宏观了解 Kettle

上一篇中对 Kettle 进行了简单的介绍，并快速体验了一把 Kettle，完成了「把数据从 CSV 文件复制到 Excel 文件」 HelloWrold 级别的功能。

而在实际工作中，可以使用 Kettle 的图形化的方式定义复杂的 ETL 程序和工作流，如下图就是通过一系列的转换（Transformation） 完成一个作业（Job）流程。

image-20210707143745802

Kettle 核心概念

image-20210708004806905

转换

转换（Transaformation）是 ETL 中最主要的部分，它处理抽取、转换、加载各种对数据行的操作。 转换包含一个或多个步骤（Step），如上图中的「CSV 文件输入」、「Excel输出」步骤，还包括过滤数据行、数据清洗、数据去重或将数据加载到数据库等等。 转换里的步骤通过跳（hop）来进行连接，跳定义一个单向通道，允许数据从一个步骤向另一个步骤流动。

步骤（Step）

image-20210708010417687

Kettle 里面的，Step 步骤是转换里的基本的组成部分，上篇快速体验的案例中就存在两个步骤，「CSV文件输入」和「Excel输出」，一个步骤有如下几个关键特性：

  • 步骤需要有一个名字，这个名字在转换范围内唯一。
  • 每个步骤都会读、写数据行（唯一例外是「生成记录」步骤，该步骤只写数据）。
  • 步骤将数据写到与之相连的一个或多个输出跳，再传送到跳的另一端的步骤。
  • 大多数的步骤都可以有多个输出跳，当有多个输出时，会弹出如下图所示的警告进行选择分发还是复制。一个步骤的数据发送可以被设置为分发和复制，分发是目标步骤轮流接收记录，复制是所有的记录被同时发送到所有的目标步骤。

image-20210708010916460

image-20210708011405544

跳（Hop）

image-20210708011632619

Kettle 里面的，跳（Hop），跳就是步骤之间带箭头的连线，跳定义了步骤之间的数据通路，如上图。在 Kettle里，数据的单位是行，数据流就是数据行从一个步骤到另一个步骤的移动， 跳是两个步骤之间的被称之为行集的数据行缓存（行集的大小可以在转换的设置里定义，如下图）。当行集满了，向行集写数据的步骤将停止写入，直到行集里又有了空间；当行集空了，从行集读取数据的步骤停止读取，直到行集里又有可读的数据行。

行集设置

数据行

在 Kettle 里，数据的单位是行，数据以数据行的形式沿着步骤移动。一个数据行是零到多个字段的集合，字段包含下面几种数据类型。

  • String：字符类型数据
  • Number:双精度浮点数
  • Integer：带符号长整型（64位）
  • BigNumber：任意精度数据
  • Date：带毫秒精度的日期时间值
  • Boolean：取值为 true 和 false 的布尔值
  • Binary：二进制字段可以包含图像、声音、视频及其他类型的二进制数据

image-20210708013817323

同时，每个步骤在输出数据行时都有对字段的描述，这种描述就是数据行的元数据。通常包含下面一些信息：

  • 名称：行里的字段名应用是唯一的
  • 数据类型：字段的数据类型
  • 格式：数据显示的方式，如 Integer 的#、0.00
  • 长度：字符串的长度或者 BigNumber 类型的长度
  • 精度：BigNumber 数据类型的十进制精度
  • 货币符号：￥
  • 小数点符号：十进制数据的小数点格式
  • 分组符号：数值类型数据的分组符号

步骤是并行的

这种基于行集缓存的规则（前面 「跳（Hop）」节提到），允许每个步骤都是由一个独立的线程运行，这样并发程度最高。这一规则也允许数据以最小消耗内存的数据流的方式来处理（设置合理的行集大小）。在数据仓库建设过程中，经常要处理大量数据，所以这种并发低消耗内存的方式也是 ETL 工具的核心需求。

对于 Kettle 的转换，所有步骤都以并发方式执行，即：当转换启动后，所有步骤都同时启动，从它们的输入跳中读取数据，并把处理过的数据写到输入跳，直到输入跳里不再有数据，就中止步骤的运行。当所有的步骤都中止了，整个转换就中止了。

总结

  • Kettle 通过一系列的转换（Transformation） 完成一个作业（Job）流程
  • 通过了解 Kettle 的核心概念，得知 Kettle 是通过「跳（Hop）」将数据流从一个步骤到另一个步骤的移动，每个步骤都是由一个独立的线程运行，这样提高并发程度，但相比 Hadoop 生态移动计算模型更加昂贵
  • Kettle 本身由 Java 开发，需要配置合理的 JVM 参数

欢迎关注公众号：HelloTech，获取更多内容

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

作者 | TCeason 青云科技数据库研发工程师

2000 年至今，MySQL[1] 一直是全球最受欢迎的 OLTP（联机事务处理）数据库，ClickHouse[2] 则是近年来受到高度关注的 OLAP（联机分析处理）数据库。那么二者之间是否会碰撞出什么火花呢？

本文将带领大家打破异构数据库壁垒，将 MySQL 数据同步至 ClickHouse。对QingCloud MySQL Plus[3] 平台与 MaterializeMySQL[4] 引擎进行了详细介绍，并进一步简述了 HTAP 应用场景。

背景

1、MySQL 复制的发展历程

file

图 1-1 MySQL 复制的发展历程
图1-1详细罗列了 MySQL 复制的发展历程。

2001 年的 MySQL 3.23 版本就已经支持了同构数据库异步复制；由于是异步复制，根本无法在实际生产中大批量使用。

2013 年 MySQL 5.7.2 版本支持增强半同步复制能力，才勉强算得上是企业级可用的数据同步方案。

2016 年 MySQL 5.7.17 支持了 MGR，并不断地发展成熟，变成了一个金融级别可用的数据同步方案。

而对于同构的 MySQL 数据同步，接下来要做的就是不断地优化体验，提升同步时效性，解决网络异常下的各类问题。

基于此，各大厂商也开始做自己的高可用同步组件。例如 QingCloud MySQL Plus，就具备了真正的强一致性和高可用能力。

2、QingCloud MySQL Plus

file

图 1-2 MySQL Plus 架构图
图 1-2 中的 Xenon 是由类 Raft 算法来实现的高可用组件，用来管理 MySQL 选举和探活，并订正数据准确性。MySQL 数据同步则依然使用 Semi-Sync Replication 或者 MGR，从而达到数据强一致性、无中心化自动选主且主从秒级切换，以及依托于云的跨区容灾能力。

多副本同步复制，确保金融级强一致性

QingCloud MySQL Plus 采用一主两从的初始节点架构设计，并通过 MySQL 5.7 版本中的 Semi-Sync 特性实现数据的多副本同步复制，确保至少一个从节点与主节点始终保持数据的完全一致，提供金融级数据强一致性。多个从节点的设置将极大地屏蔽掉单点故障带来的影响，确保集群内始终有从节点保有全量数据。

无中心化自动选主且主从秒级切换

节点之间使用 Raft 协议进行管理，当主节点出现故障不可用时，集群会秒级响应并选出新的主节点（与主节点数据完全同步的从节点），立即接管读写请求，确保业务的连续高可用。这一过程，用户完全无需关心后端集群中各节点的角色如何设置，一切由系统自动管理。

跨区容灾能力

可实现多可用区主从部署，具有跨可用区容灾能力，提升数据安全性及容灾能力。

MySQL 有了高可用能力之后，可以通过增加只读实例的方式来增强 AP 能力。但是 MySQL 数据结构和分布方式决定了其 AP 能力相对较弱，那么如何加速 OLAP 查询呢？

ClickHouse 同步 MySQL 数据

为了加速 OLAP 查询，QingCloud MySQL Plus 借用 ClickHouse 来同步 MySQL 数据。

1、ClickHouse 概述

file

图2-1 ClickHouse 产品图
ClickHouse 是一个用于联机分析 (OLAP) 的列式数据库管理系统 (DBMS)。ClickHouse 构思于 2008 年，最初是为 YandexMetrica（世界第二大Web分析平台）而开发的。多年来一直作为该系统的核心组件被该系统持续使用着，并于 2016 年宣布开源。

file

图 2-2 ClickHouse 热度趋势图
从目前最新的 DB-Engines 中可以看到其排名曲线一路高涨，并且各大厂在重要业务上已经大量部署，这是一个很明显的趋势。因此，我们似乎可以认定 ClickHouse 的火热并不只是一时现象，它将长久地存活下去。而且，ClickHouse 灵活的外部表引擎，可轻松实现与 MySQL 的数据同步，接下来让我们了解一下。

2、MySQL Table Engine

MySQL Table Engine 的特性

  • Mapping to MySQL table
  • Fetch table struct from MySQL
  • Fetch data from MySQL when executing query

ClickHouse 最开始支持表级别同步 MySQL 数据，通过外部表引擎 MySQL Table Engine 来实现同 MySQL 表的映射。从 MySQL 的 information_schema 中获取对应表的结构，将其转换为 ClickHouse 支持的数据结构，此时在 ClickHouse 端，表结构建立成功。但是此时，并没有真正去同步数据。只有向 ClickHouse 中的该表发起请求时，才会主动的拉取要同步的 MySQL 表的数据。

MySQL Table Engine 使用起来非常简陋，但它是非常有意义的。因为这是第一次打通 ClickHouse 和 MySQL 的数据通道。但是，缺点异常明显：

  1. 仅仅是对 MySQL 表关系的映射；
  2. 查询时传输 MySQL 数据到 ClickHouse，会给 MySQL 可能造成未知的网络压力和读压力，可能影响 MySQL 在生产中正常使用。

基于 MySQL Table Engine 只能映射 MySQL 表关系的缺点，QingCloud ClickHouse 团队实现了MySQL Database Engine。

3、MySQL Database Engine

MySQL Database Engine 的特性

  • Mapping to MySQL Database
  • Fetch table list from MySQL
  • Fetch table struct from MySQL
  • Fetch data from MySQL when executing query

MySQL Database Engine 是库级别的映射，依然要从 information_schema 中拉取待同步库中包含的所有 MySQL 表的结构，解决了需要建立多表的问题。但仍然还有和 MySQL Table Engine 一样的缺点：查询时传输 MySQL 数据到 ClickHouse，给 MySQL 可能造成未知的网络压力和读压力，可能影响 MySQL 在生产中正常使用。

4、借用第三方软件同步

file

图 2-3 借用第三方软件同步数据
除去上面提到的 MySQL Table Engine 、MySQL Database Engine 两种方式，还有可以采用第三方软件来同步数据，比如 Canal 或者 Kafka，通过解析 MySQL binlog，然后编写程序控制向 ClickHouse 写入。这样做有很大的优势，即同步流程自主可控。但是也带来了额外的问题：

  1. 增加了数据同步的复杂度。
  2. 增加了第三方软件，使得运维难度指数级增加。

基于此，我们又可以思考一个问题，ClickHouse 能否主动同步并订阅 MySQL 数据呢？

Materialize MySQL

为了解决 MySQL Database Engine 依然存留的问题，支持 ClickHouse 主动同步并订阅 MySQL 数据，QingCloud ClickHouse 团队自主研发了 MaterializeMySQL 引擎。

1、简述 MaterializeMySQL

MaterializeMySQL 引擎是由 QingCloud ClickHouse 团队自主研发的库引擎，目前作为实验特性合并到 ClickHouse 20.8 版本中，是对 MySQL 库级别关系的映射，通过消费 binlog 存储到 MergeTree 的方式来订阅 MySQL 数据。

1
2
3
bash复制代码CREATE DATABASE test ENGINE = MaterializeMySQL(
  '172.17.0.3:3306', 'demo', 'root', '123'
)

具体使用方式就是一条简单的 CREATE DATABASE SQL 示例：

172.17.0.3:3306 - MySQL 地址和端口
demo - MySQL 库的名称
root - MySQL 同步账户
123 - MySQL 同步账户的密码

2、MaterializeMySQL 的设计思路

  • Check MySQL Vars
  • Select history data
  • Consume new data

MaterializeMySQL 的设计思路如下：

a. 首先检验源端 MySQL 参数是否符合规范。
b. 再将数据根据 GTID 分割为历史数据和增量数据。
c. 同步历史数据至 GTID 点。
d. 持续消费增量数据。

3、MaterializeMySQL 的函数流程

file

图3-1 MaterializeMySQL 函数流程
如图3-1 所示，MaterializeMySQL 函数的主体流程为：

CheckMySQLVars -> prepareSynchronized -> Synchronized

（1）CheckMySQLVars

检验参数比较简单，就是要查询这些参数是否符合预期。

1
2
3
4
5
6
7
8
9
10
sql复制代码SHOW VARIABLES WHERE (Variable_name = 'log_bin'
AND upper(Value) = 'ON')
OR (Variable_name = 'binlog_format'
AND upper(Value) = 'ROW')
OR (Variable_name = 'binlog_row_image'
AND upper(Value) = 'FULL')
OR (Variable_name = 'default_authentication_plugin'
AND upper(Value) = 'MYSQL_NATIVE_PASSWORD')
OR (Variable_name = 'log_bin_use_v1_row_events'
AND upper(Value) = 'OFF');

（2）prepareSynchronized

这一步来实现历史数据的拉取。

  • 先初始化 gtid 信息。
  • 为了保证幂等性每次重新同步时，都要清理 ClickHouse MaterializeMySQL 引擎库下的表。
  • 重新拉取历史数据，并将 MySQL 表结构在 ClickHouse 端进行改写。
  • 建立与 MySQL 的 Binlog 传输通道。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
scss复制代码std::optional<MaterializeMetadata> MaterializeMySQLSyncThread::prepareSynchronized()
{
    connection = pool.get();
    MaterializeMetadata metadata(
connection, DatabaseCatalog::instance().getDatabase(database_name)->getMetadataPath() + "/.metadata", mysql_database_name, opened_transaction);
    if (!metadata.need_dumping_tables.empty())
    {
        Position position;
        position.update(metadata.binlog_position, metadata.binlog_file, metadata.executed_gtid_set);
        metadata.transaction(position, [&]()
        {
            cleanOutdatedTables(database_name, global_context);
            dumpDataForTables(connection, metadata, query_prefix, database_name, mysql_database_name, global_context, [this] { return isCancelled(); });
         });
    }
    connection->query("COMMIT").execute();
}

在 MySQL 中，demo 库下有一个表 t ，主键为 ID , 普通列 col_1。

1
2
3
4
5
sql复制代码CREATE TABLE demo.t (
  id int(11) NOT NULL,
  col_1 varchar(20) DEFAULT NULL,
  PRIMARY KEY (id)
) ENGINE = InnoDB;

在 ClickHouse 中，依然是 id 作为主键列，但是，多了隐藏列 _sign 和 _version。

  1. _sign：值只有 1 和 -1。其中，1 代表这行数据存在，-1 代表这行数据被删除。
  2. _version：只会读到 version 高的值，会在后台不断合并主键相同的行，最终保留 Version 最高的行。
1
2
3
4
5
6
7
8
9
10
scss复制代码CREATE TABLE test.t
(
    `id` Int32,
    `col_1` Nullable(String),
    `_sign` Int8 MATERIALIZED 1,
    `_version` UInt64 MATERIALIZED 1
)
ENGINE = ReplacingMergeTree(_version)
PARTITION BY intDiv(id, 4294967)
ORDER BY tuple(id)

（3）Synchronized

在 prepareSynchronized 中，我们得到了历史数据以及历史数据位点信息，并且获得了与 MySQL 的 Binlog 传输通道。接下来就是从该位点同步增量数据。通过 readOneBinlogEvent 函数读取每一条 binlog 内容，然后使用 onEvent 转换成 ClickHouse 的语句格式即可。最终为了数据安全性，调用 flushBuffersData 函数将数据落盘。

1
2
3
4
5
6
7
8
9
10
11
scss复制代码client.connect();
client.startBinlogDumpGTID(randomNumber(), mysql_database_name, metadata.executed_gtid_set, metadata.binlog_checksum);
Buffers buffers(database_name);
while (!isCancelled())
{
    BinlogEventPtr binlog_event = client.readOneBinlogEvent(std::max(UInt64(1), max_flush_time - watch.elapsedMilliseconds()));
    if (binlog_event)
        onEvent(buffers, binlog_event, *metadata);
    if (!buffers.data.empty())
        flushBuffersData(buffers, *metadata);
}

HTAP 应用场景

file

图 4-1 Materialize 实现 HTAP 架构图
当我们打通了 ClickHouse 和 MySQL 的复制通道，而 ClickHouse 的分析能力又是如此让人惊喜，那么我们是不是可以用 MySQL Plus + ClickHouse 实现 HTAP 呢？

在图 4-1 中的架构，依然使用高可用组件 Xenon 来管理 MySQL 复制，同时 Xenon 增加了对 ClickHouse 的监管，通过 MaterializeMySQL 来同步 MySQL 数据。

在之前的 MySQL Plus 架构图中，使用 MySQL 只读实例来进行商务分析、用户画像等分析业务。而现在可以直接将 ClickHouse 作为一个分析实例加入到 MySQL 复制中，替代一部分只读实例进行分析计算。同时 ClickHouse 本身支持了海量函数来支持分析能力的同时还支持标准 SQL，相信可以让使用者享受到很好的体验。

目前的 ClickHouse 可以支持同步 MySQL 5.7 和 8.0 的数据，不支持同步 MySQL 5.6 的数据。不过，作为一个实验特性， MaterializeMySQL 的时间线相当于是 2001 年刚刚支持复制的 MySQL。欢迎大家一起来贡献和维护 MaterializeMySQL。

[1]. MySQL : www.mysql.com/

[2]. ClickHouse : clickhouse.tech/docs/en/

[3]. MySQL Plus：www.qingcloud.com/products/my…

[4]. MaterializeMySQL：clickhouse.tech/docs/en/eng…

本文由博客一文多发平台 OpenWrite 发布！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

最近想要打通几个应用程序的用户关系，搞一个集中式的用户管理系统来统一管理应用的用户体系。经过一番调研选中了红帽开源的Keycloak，这是一款非常强大的统一认证授权管理平台。之所以选中了Keycloak是基于以下几个原因。

易用性

Keycloak为Web应用和Restful服务提供了一站式的单点登录解决方案。它的目标就是让应用的安全管理变得简单，让开发人员可以轻松地保护他们的应用程序和服务。并且Keycloak为登录、注册、用户管理提供了可视化管理界面，你可以借助于该界面来配置符合你需要的安全策略和进行用户管理。 而且还可以

登录界面

可配置的GUI管理

功能强大

Keycloak实现了业内常见的认证授权协议和通用的安全技术，主要有：

  • 浏览器应用程序的单点登录（SSO）。
  • OIDC认证授权。
  • OAuth 2.0。
  • SAML。
  • 多租户支持。
  • 身份代理 - 使用外部 OpenID Connect 或 SAML 身份提供商进行身份验证。
  • 第三方登录。
  • 用户联盟 - 从 LDAP 和 Active Directory 服务器同步用户。
  • Kerberos 网桥 - 自动验证登录到 Kerberos 服务器的用户。
  • 用于集中管理用户、角色、角色映射、客户端和配置的管理控制台。
  • 用户账户集中管理的管理控制台。
  • 自定义主题。
  • 两段身份认证。
  • 完整登录流程 - 可选的用户自注册、恢复密码、验证电子邮件、要求密码更新等。
  • 会话管理 - 管理员和用户自己可以查看和管理用户会话。
  • 令牌映射 - 将用户属性、角色等映射到令牌和语句中。
  • 安全策略恢复功能。
  • CORS 支持 - 客户端适配器具有对 CORS 的内置支持。
  • 自定义SPI接口扩展。
  • JavaScript 应用程序、WildFly、JBoss EAP、Fuse、Tomcat、Jetty、Spring 等客户端适配器。
  • 支持任何具有 OpenID Connect Relying Party 库或 SAML 2.0 Service Provider 库的平台/语言。

而且有专门的Spring Boot Starter，非常容易集成到Spring Boot中。

基于实践的开源

红帽出品，必属精品。红帽良好的口碑决定了Keycloak的可靠性。它遵循Apache 2.0开源协议进行开源，经过八年的持续开源，代码质量很高，非常适合做定制化开发。红帽的商业付费认证授权产品Red Hat SSO就是基于Keycloak。为企业提供了动态单点登录的解决方案，间接证明了Keycloak的可靠性。

适配Spring Security

这个框架对Spring Security和Spring Boot做了适配，非常适合使用了这两种体系的迁移扩展。这也是我选择它的重要原因之一。

缺点

虽然优点非常多，但是缺点也很明显。功能强大就意味着架构比较复杂，概念比较多，学习成本比较高。

KeyClock的核心概念

学习成本比较高的另一个原因是中文资料比较少，需要自己去啃官方的文档。对于业务需要的认证方式可能会需要自行实现一些接口，同样考验着个人的编码能力。

最后

胖哥对这个东西关注了很久却没有下手，第一是因为它确实有挑战性，第二没有实际的开发场景，现在机会来了，今天对这个框架进行一个简单的介绍，让不了解它的同学先简单了解一下。如果你对Keycloak进行了详细的研究和实践，基本上能够搞定一些大中型的应用安全体系构建，既有诱惑也有挑战。另外这个程序适合做统一认证授权门户构建，不太适合一些小应用，相对比较重，不过微服务用这个应该非常棒。在目前新的Spring认证服务器还没有达到生产可用时是一个不错的选择。所以后续会和大家一起研究学习这个东西，感兴趣的朋友可以多多关注：码农小胖哥。

关注公众号：Felordcn获取更多资讯

个人博客：https://felord.cn

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

欢迎来到《王者并发课》，本文是该系列文章中的第22篇，铂金中的第9篇。

在前面的文章中，我们已经介绍了ReentrantLock，CountDownLatch，CyclicBarrier，Semaphore等同步工具。在本文中，将为你介绍最后一个同步工具，即Exchanger.

Exchanger用于两个线程在某个节点时进行数据交换。在用法上，Exchanger并不复杂，但是实现上会稍微有点费解。所以，考虑到Exchanger在平时使用的场景并不多，况且多数读者对一些“枯燥”的源码的耐受度有限（可能引起不适或烦躁等不良情绪，阻碍学习），本文将侧重讲它的使用和思想，对于源码不会过多展开，点到为止。

一、Exchanger的使用场景

在峡谷中，铠和兰陵王都是擅长打野的英雄，各自对野怪的偏好也不完全相同。所以，为了能得到自己想要的野怪，他们经常会在峡谷的交易中心交换各自的猎物。

这一天，铠打到了一只棕熊，而兰陵王则收获了一只野狼，并且彼此都想要对方的野怪。于是，他们约定在峡谷交易中心交换双方的野怪，谁先到了就先等会。这个过程，可以用下面这幅图来表示：

在铠和兰陵王交换猎物的过程中，有三个点需要你留意：

  • 交换的双方有明确的交易地点（峡谷交易中心）；
  • 交换的双方具有明确的交易对象（比如棕熊和野狼）；
  • 谁先到了就等会儿（他们中总会有先来后到）。

如果用代码来实现的话，也是有多种方式可以选择，比如前面所学过的同步方法等。不过，虽然做也是可以做的，只是没那么方便。所以，接下来我们就用Exchanger来实现这一过程。

在下面的代码中，我们定义了一个exchanger，它就类似于峡谷交易中心，而它的类型Exchanger<WildMonster> 则明确表示交换的对象是野怪。

接着，我们再定义两个线程，分别代表铠和兰陵王。在其线程的内部，会通过前面定义的exchanger对象来和对方进行交换数据。交换完成后，他们彼此将获得对方的物品。

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
java复制代码 public static void main(String[] args) {
     Exchanger<WildMonster> exchanger = new Exchanger<> (); // 定义交换地点和交换类型

     Thread 铠 = newThread("铠", () -> {
         try {
             WildMonster wildMonster = new Bear("棕熊");
             say("我手里有一只：" + wildMonster.getName());
             WildMonster exchanged = exchanger.exchange(wildMonster); // 交换后将获得对方的物品
             say("交易完成，我获得了：", wildMonster.getName(), "->", exchanged.getName());
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
     });

     Thread 兰陵王 = newThread("兰陵王", () -> {
         try {
             WildMonster wildMonster = new Wolf("野狼");
             say("我手里有一只：" + wildMonster.getName());
             WildMonster exchanged = exchanger.exchange(wildMonster);
             say("交易完成，我获得了：", wildMonster.getName(), "->", exchanged.getName());
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
     });
     铠.start();
     兰陵王.start();
 }

下面是上面代码用到的内部类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
java复制代码 @Data
 private static class WildMonster {
     protected String name;
 }

 private static class Wolf extends WildMonster {
     public Wolf(String name) {
         this.name = name;
     }
 }

 private static class Bear extends WildMonster {
     public Bear(String name) {
         this.name = name;
     }
 }

示例代码运行结果如下：

1
2
3
4
5
6
java复制代码铠:我手里有一只：棕熊
兰陵王:我手里有一只：野狼
兰陵王:交易完成，我获得了：野狼->棕熊
铠:交易完成，我获得了：棕熊->野狼

Process finished with exit code 0

从结果中可以看到，铠用棕熊换到了野狼，而兰陵王则用野狼换到了棕熊，他们完成了交换。

以上就是Exchanger的用法，看起来还是非常简单的，事实上也确实很简单。在使用Exchanger的时候要注意下面几点：

  • 定义Exchanger对象，各线程通过这个对象完成交换；
  • 在Exchanger对象中要定义类型，也就是这两个线程要交换什么；
  • 线程在调用Exchanger进行交换时，要特别注意的是，先到的那个线程会原地等待另外一个线程的出现。比如，铠先到交换地点，可这时候兰陵王还没有到，那么铠会等待兰陵王的出现，除非超过设置的时间限制，比如兰陵王中途被妲己蹲了草丛。反之亦然，兰陵王先到也到等铠的出现。

二、Exchanger的源码与实现

虽然理解Exchanger的思想很容易，了解其用法也很简单，但是若要理清它几百余行的源码却并非易事。其原因在于，槽是Exchanger中的核心概念和属性，Exchanger中的数据交换分为单槽交换和多槽交换，其中单槽交换源码简单，但多槽交换却很复杂。所以，下文对Exchanger源码的阐述以概括为主，不会对源码深究。如果你有兴趣，可以参考阅读这篇文章，作者对其源码的解读较为详细。

1. 核心构造

与其他同步工具不同的是，Exchanger有且仅有一个构造函数。在这个构造中，也只初始化了一个对象participant.

1
2
3
java复制代码public Exchanger() {
    participant = new Participant();
}

从继承关系看，Participant本质上是一个ThreadLocal，而其中的Node则是线程的本地变量。

1
2
3
4
5
java复制代码static final class Participant extends ThreadLocal<Node> {
    public Node initialValue() { 
        return new Node(); 
    }
}

2. 核心属性

Exchanger有四个核心变量，如下所示。当然，除此之外，还有一些用以计算的其他变量。不过，为避免引入不必要的复杂度，本文暂不提及。

1
2
3
4
5
6
7
8
9
10
11
java复制代码//ThreadLocal变量，每个线程都有自己的一个副本
private final Participant participant;

//多槽位，高并发下使用，保存待匹配的Node实例
private volatile Node[] arena;

//单槽位，arena未初始化时使用的保存待匹配的Node实例
private volatile Node slot;

//初始值为0，当创建arena后会被赋值成SEQ，用来记录arena数组的可用最大索引，会随着并发的增大而增大直到等于最大值FULL，会随着并行的线程逐一匹配成功而减少恢复成初始值
private volatile int bound;

Node的具体细节，注意其中的item和match.

1
2
3
4
5
6
7
8
9
java复制代码 @sun.misc.Contended static final class Node {
        int index;              //arena的下标，多个槽位的时候使用
        int bound;              // 上一次记录的Exchanger.bound
        int collides;           // 记录的 CAS 失败数
        int hash;               // 用于自旋
        Object item;            //  这个线程的数据项
        volatile Object match;  // 交换的数据
        volatile Thread parked; //  当阻塞时，设置此线程，不阻塞的话会自旋
    }

3. 核心方法

1
2
3
4
5
6
7
java复制代码// 交换数据
// 如果一个线程达到后，会等待其他线程的到达（除非自己被中断）。然后，该线程会和到达的线程交换数据。
// 如果线程在到达后，已经有其他线程在等待。那么，将会唤起该线程并交换数据。
public V exchange(V x) throws InterruptedException {...}
    
//带有超时限制的交换
public V exchange(V x, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {...}

所以，从源码上看上文的示例，那么铠和兰陵王交换数据的过程应该是下面这样的：

小结

以上就是关于Exchanger的全部内容。在学习Exchanger时，要侧重理解它所要解决的问题场景，以及它的基本用法。对于其源码，当前阶段可以选择“不求甚解”，以降维的方式降低学习难度，日后再循序渐进理解。我在写本文时，也曾多次考虑是否要讲清楚源码，最终还是决定暂缓，毕竟现阶段理解它、学会它才是重点。

正文到此结束，恭喜你又上了一颗星✨

夫子的试炼

  • 使用Exchanger实现生产者与消费者。

延伸阅读与参考资料

关于作者

从业近十年，先后从事敏捷与DevOps咨询、Tech Leader和管理等工作，对分布式高并发架构有丰富的实战经验。热衷于技术分享和特定领域书籍翻译，掘金小册《高并发秒杀的设计精要与实现》作者。

关注公众号【MetaThoughts】，及时获取文章更新和文稿。

如果本文对你有帮助，欢迎点赞、关注、监督，我们一起从青铜到王者。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

Dbeaver是一个图形化的界面工具，专门用于与各种数据库的集成，通过dbeaver我们可以与各种数据库进行集成。通过图形化界面的方式来操作我们的数据库与数据库表，类似于我们的sqlyog或者navicat。

下载Dbeaver

常见报错

1
2
3
4
5
perl复制代码报错信息：
Could not establish connection to jdbc:hive2://172.16.250.240:10000/test: Required field 'client_protocol' is unset! Struct:TOpenSessionReq(client_protocol:null, configuration:{use:database=test})

Required field 'client_protocol' is unset! Struct:TOpenSessionReq(client_protocol:null, configuration:{use:database=test})
Required field 'client_protocol' is unset! Struct:TOpenSessionReq(client_protocol:null, configuration:{use:database=test})

安装dbeaver

根据下载的dbeaver.dmg，双击安装

然后启动dbeaver图形化界面

新建hive连接，配置主机名、端口号、用户名等

配置修改驱动

1
2
3
ruby复制代码下载需要的驱动
[hadoop@node03 ~]$ cd /opt/module/hive-1.1.0-cdh5.14.2/lib/
[hadoop@node03 lib]$ sz hive-jdbc-1.1.0-cdh5.14.2-standalone.jar

删除默认驱动配置

加载下载的驱动配置

测试连接

Dbeaver解决无法下载数据库驱动问题解决

Dbeaver专门用于与各种数据库的集成，通过Dbeaver我们可以与各种数据库进行集成，不同的数据库类型连接其需要下载不同的驱动，往往驱动包都是国外，因为都知道的原因会出现下载慢和无法下载驱动的问题，下面整理了解决驱动下载问题的方法。

配置hive数据库连接，增加相关配置

点击测试连接按钮，此时会下载驱动，往往就在此时出现驱动慢或者无法下载问题。

在Hive连接配置页面，编辑驱动设置

修改驱动所使用的的maven镜像为阿里云源的maven

1
ruby复制代码alimaven aliyun maven http://maven.aliyun.com/nexus/content/groups/public/ central

驱动下载完成，测试数据库链接

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

大家好，我是bigsai，今天给大家分享自己字节跳动面试经验分享。

enum我面得岗位是后台实习开发，具体部门是懂车帝，总体感觉就是字节的流程真的好快，只要安排面试，那流程接着很快。

大概是上上周投递一波简历，然后上周一约的面试，然后周二下午三场面试(腾讯三面) 然后字节一面完面试官说你休息两分钟，待会二面，然后就接着二面，周三就是hr面，周五具体offer就下来了是真的快。

字节的两次面试大概每次都是一个小时左右，面试主要内容就是 算法题+项目+八股文，下面会详细讲讲自己的看法。

字节一面 1h

ps:字节的后台主要技术栈是Golang，但是一面面试官看我简历上是Java技术栈，并且这个老哥感觉以前是做Java的，所以问了一些Java的问题。

自我介绍

算法题(剑指offer32 从上到下打印二叉树 III)

介绍一下项目

在项目中负责什么(问数据库设计、缓存设计、其他功能设计等等)

你是搞Java的是吧，说说Spring两大核心(IOC,AOP)

SpringBoot和Spring区别

JVM内存模型

讲讲你知道的锁，什么是可重入锁

ReentrantLook是怎么实现

讲讲RESTful

讲讲InnerDB

说说你知道的HTTP状态码

Nginx和Tomcat区别

Get和Post的区别

HTTP请求报文格式

三次握手四次挥手

如何设计一个登录功能，需要考虑什么东西

字节二面(接一面)1h

自我介绍

算法题(力扣56. 合并区间)

介绍一下项目

项目中有用到什么设计模式，介绍你熟悉的设计模式

进程通信有哪几种方式，线程通信有哪几种方式

浏览器输入一个URL过程

MySQL事务ACID

事务隔离级别、幻读

谈谈你了解的微服务

流量并发太大有什么优化方法

Session存储在哪里，和Cookie区别

谈谈消息队列

如何保证Web应用设计的安全性(想多少说多少)

场景题：设计一个评论系统(数据库、缓存最重要)，评论可以curd，查询可以按照页面ID查询，和用户ID查询(可以简单说说可以详细说说，自由发挥)

ps:这个场景题很有意思，缓存设计我个人设计前几页热点评论放缓存，其他不放(因为有删除可能影响太大hhh)

HR面

自我介绍一下

在项目中承担什么角色

你有什么优势

……

什么时候能入职

过两天offer就下来了

image-20210705180040946

总结

enum，因为是实习面试，可能没那么卷，个人感觉大部分问题并没有刁难，也并没有深挖一两个点，刚开始我以为字节会以恐怖算法题和八股文让我整个过程瑟瑟发抖，我以为疯狂计网、疯狂OS，但是两个面试官都跟我聊了很多项目相关的，并且我的项目都是以前自己负责编写，也参加过一些比赛啥的，所以在项目润色这块可谓是很早前就做了准备。

根据刷题，八股文，和项目在这里我给大家一些建议吧。

刷题

刷题的话，短期速成很难，不过实习大部分公司对刷题要求没正式批那么高(毕竟是进去廉价劳动力)，并且刷题实习一般不会出hard难度的问题，除非面试官可能看你不爽(hhh)，然后剑指offer优先级最高，把剑指offer简单和中等题目认真刷，最好是懂每一题的最佳解 ，不过有时在面试途中如果直接用最佳方法怕卡壳忘记思路断电(确实有这种情况)，你可以用自己把握强的方法先过，然后跟面试官说：”面试官这题虽然这样过了，我还有优化方法，可以跟你说说吗。” 然后就这样可以再把最佳方法思路流程详细说一下。

这个当场刷题，其实考的就是一个刷题量，简单题你可能当场想出来，但是剑指offer，力扣这些基本都是思维技巧很大的当场这种紧张很容易断电，所以还是要提前准备，面对字节实习的话最起码要准备剑指offer和力扣hot100题。

可不能在算法题直接倒下，在这倒下很影响你后面回答问题信心，并且面试官如果看你题解答的很棒可能用一个交流的方式跟你面试，而不是冷冰冰的一直问一直挖哈哈(个人猜测)。

八股文

八股文因为我准备时间很少，可能只有有限天数，我还是推荐Guide哥的JavaGuide面试突击版，用起来感觉很香，把很多高频问题都已经总结到了，另外如果有些感觉如果不够详细，可以找一些博客文章具体看看，因为面试高频问题有很多人写的非常好。不过面字节和腾讯的话Java那部分可以省略跳过一下。

除此之外，看一些公众号文章也是有必要的，比如我面试前把秀哥逆袭进大厂系列需要的看了，还有狼王那篇GET和POST卡布奇诺也看了刚好也问到了，还有看了一些身边比较厉害号主文章感觉还是在面试上有不少帮助的。

最后一点，牛客面经肯定要看的，看看各个大厂(可以直接搜这个部门)常见面试问题，看看自己能不能回答上来，不会的就去搜索学习一下，虽然这个很零散，但是经过前面两个步骤大部分热门面试题可能都涵盖了，所以再看看面经就是一方面考察一下自己掌握情况，另一方面就是查漏补缺。

项目

以前听过很多人靠着刷题+八股文就能进大厂，但是最近我感觉各个大厂对项目的要求明显高了，会详细问项目中的各种实现，也会给各种场景，所以要好好准备一些项目，最好是自己参与的。

项目最好什么类型呢？不要商城系统、秒杀系统、图书管理系统这些，最好就是能贴合以下两个方面：

  1. 能够真实落地小部分范围使用的，在这个过程中肯定有一个发展的过程。
  2. 未落地但是有发展前景的，跟当今潮流关系比较大的方向比如教育之类(可以多吹)。

项目的话一般准备2-3个，我个人准备两个，但是两个项目最好类别有一定区分度，最好不是课设、毕设这种必须要求的项目而是自己更主动创造出来的(可以说是参赛项目、落地使用项目、开源项目等等)，并且项目中基本覆盖主流技术栈，这样面试官可以通过项目问一些问题你可能更熟悉，回答的也更好。

项目的话最好是自己的，如果自己没有项目也没有精力写项目可以找，但是不是自己的话一定保证吃透，可以使学长、学弟的，可以跟他们交流一下，吃不透面试官一问开发中具体可能出现问题如果露馅那可能评价不太好。

其他

除了上面提到的，简历啦这些可能也很重要，简历主要是包含基本信息、比赛荣誉、项目经历、实习经历(部分无)、个人评价这几个部分，最好各个部分充实，排版简洁(不过各大官网会有他们自己简历系统)。与面试官交流要口齿清楚，与人良好沟通也是实习工作必备条件，面试途中要谦虚乐观，这些不用多说了。

各个方面趁早准备，很多东西身不由己，计划早一点总没坏处。

最后

这其实是我收获的第一个大厂offer，虽然是实习的，考研狗一直在校读书到了研一读了五年书没实习过确实是有些遗憾，给hr交流时候喊她姐她让我别喊姐，我翻了朋友圈才发现她也是hr实习生hhh难怪。

但是这个offer我还是拒了，本来打算这个暑假大厂实习一波，顺便看看论文，然后秋天可能回南京本部搞搞论文。

但hhh研究生其实就是打工人，干活。暑期可能外派28所和大西北参加无人车项目，不过这无人车项目是c语言的，跟就业无关。我们学校并且很多学校的研究生可能都这样：研一上课、看论文、有的干干活，研二要写论文、干活，到了找工作的时候发现自己好像什么都不太会，从0开始学，然后最后只能去华为、国企这些更吃学历不太吃技术的地方。

image-20210705175538596

image-20210705175708530

虽然每个人都有解不开的枷锁，但是有时候也只能认了，也只能夹缝中寻求求生的方法。不过我还好点，在江阴呆了一年很自在很轻松，比起其他被压迫同学来说还是很幸运的。

最后欢迎有需要的和我交流，如果有需要看简历、交流的欢迎加我微信(bigsai66)交流哇。

后面可能要去大西北肝文了，hhh兄弟们三连支持一下！

微信搜一搜【bigsai】关注我，获取更多干货！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

前言

最近开始阅读redis源码, 工欲善其事，必先利其器, 首先学习如果使用Clion调试Redis的源码

环境

  • MacOs 15
  • redis版本: 6.0.14
  • Clion

下载Redis源码

地址: github.com/redis/redis…
可以直接下载安装包解压即可

编辑CMakeList.txt

  1. 打开Clion, 选择: File => New CMake Project from Sources, 打开源码项目, 会自动生成根目录下的CMakeList.txt 文件
  2. 编辑一下 CMakeList.txt 内容, 没有创建即可:
1
2
3
4
5
objectivec复制代码./CMakeLists.txt
./src/modules/CMakeLists.txt
./deps/CMakeLists.txt
./deps/linenoise/CMakeLists.txt
./deps/lua/CMakeLists.txt

其中个文件内容如下:

  • ./CMakeLists.txt
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
bash复制代码cmake_minimum_required(VERSION 3.0 FATAL_ERROR)
project(redis VERSION 6.0)
#set(CMAKE_INSTALL_PREFIX "${CMAKE_BINARY_DIR}/")
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/../src)
message(CMAKE_RUNTIME_OUTPUT_DIRECTORY is:${CMAKE_RUNTIME_OUTPUT_DIRECTORY})
#if (NOT CMAKE_BUILD_TYPE)
message(STATUS "No build type defined; defaulting to 'Debug'")
set(CMAKE_BUILD_TYPE "Debug" CACHE STRING
        "The type of build. Possible values are: Debug, Release,
RelWithDebInfo and MinSizeRel.")

#endif()
message(STATUS "Host is: ${CMAKE_HOST_SYSTEM}. Build target is:
${CMAKE_SYSTEM}")
get_filename_component(REDIS_ROOT "${CMAKE_CURRENT_SOURCE_DIR}" ABSOLUTE)
message(STATUS "Project root directory is: ${REDIS_ROOT}")
# Just for debugging when handling a new platform.
if (false)
    message("C++ compiler supports these language features:")
    foreach (i ${CMAKE_CXX_COMPILE_FEATURES})
        message(" ${i}")
    endforeach ()
endif ()
message(STATUS "Generating release.h...")
execute_process(
        COMMAND sh -c ./mkreleasehdr.sh
        WORKING_DIRECTORY ${REDIS_ROOT}/src/
)
add_subdirectory(deps)
add_subdirectory(src/modules)
set(SRC_SERVER_TMP
        src/crcspeed.c
        src/crcspeed.h
        src/sha256.c
        src/sha256.h
        src/connection.c
        src/connection.h
        src/acl.c
        src/timeout.c
        src/tracking.c
        src/tls.c
        src/adlist.c
        src/ae.c
        src/anet.c
        /usr/local/include/event.h
        src/ae_kqueue.c
        src/dict.c
        src/sds.c
        src/zmalloc.c
        src/lzf_c.c
        src/lzf_d.c
        src/pqsort.c
        src/zipmap.c
        src/sha1.c
        src/ziplist.c
        src/release.c
        src/networking.c
        src/mt19937-64.c
        src/util.c
        src/object.c
        src/db.c
        src/replication.c
        src/rdb.c
        src/t_string.c
        src/t_list.c
        src/t_set.c
        src/t_zset.c
        src/evict.c
        src/defrag.c
        src/module.c
        src/quicklist.c
        src/expire.c
        src/childinfo.c
        src/redis-check-aof.c
        src/redis-check-rdb.c
        src/lazyfree.c
        src/geohash.c
        src/rax.c
        src/geohash_helper.c
        src/siphash.c
        src/geo.c
        src/t_hash.c
        src/config.c
        src/aof.c
        src/pubsub.c
        src/multi.c
        src/debug.c
        src/sort.c
        src/intset.c
        src/syncio.c
        src/cluster.c
        src/crc16.c
        src/endianconv.c
        src/slowlog.c
        src/scripting.c
        src/bio.c
        src/rio.c
        src/rand.c
        src/memtest.c
        src/crc64.c
        src/bitops.c
        src/sentinel.c
        src/notify.c
        src/setproctitle.c
        src/blocked.c
        src/hyperloglog.c
        src/latency.c
        src/sparkline.c
        src/t_stream.c
        src/lolwut.c
        src/lolwut.h
        src/lolwut5.c
        src/lolwut6.c
        src/listpack.c
        src/localtime.c
        src/gopher.c
        )
set(SRC_SERVER src/server.c ${SRC_SERVER_TMP})
set(SRC_CLI
        src/anet.c
        src/sds.c
        src/adlist.c
        src/redis-cli.c
        src/zmalloc.c
        src/release.c
        src/ae.c
        src/crc64.c
        src/crc16.c
        src/dict.c
        src/siphash.c
        )
if (${CMAKE_SYSTEM_NAME} MATCHES "Linux")
    # better not to work with jemalloc
endif()
set(EXECUTABLE_OUTPUT_PATH src)
add_executable(redis-server ${SRC_SERVER})
add_executable(redis-cli ${SRC_CLI})
set_property(TARGET redis-server PROPERTY C_STANDARD 99)
set_property(TARGET redis-server PROPERTY CXX_STANDARD 11)
set_property(TARGET redis-server PROPERTY CXX_STANDARD_REQUIRED ON)
set_property(TARGET redis-cli PROPERTY C_STANDARD 99)
set_property(TARGET redis-cli PROPERTY CXX_STANDARD 11)
set_property(TARGET redis-cli PROPERTY CXX_STANDARD_REQUIRED ON)
target_include_directories(redis-server
        PRIVATE ${REDIS_ROOT}/deps/hiredis
        PRIVATE ${REDIS_ROOT}/deps/linenoise
        PRIVATE ${REDIS_ROOT}/deps/lua/src
        )
target_include_directories(redis-cli
        PRIVATE ${REDIS_ROOT}/deps/hiredis
        PRIVATE ${REDIS_ROOT}/deps/linenoise
        PRIVATE ${REDIS_ROOT}/deps/lua/src
        )
target_link_libraries(redis-server
        PRIVATE pthread
        PRIVATE m
        PRIVATE lua
        PRIVATE linenoise
        PRIVATE hiredis
        )
target_link_libraries(redis-cli
        PRIVATE pthread
        PRIVATE m
        PRIVATE linenoise
        PRIVATE hiredis
        )
link_directories(deps/hiredis/ deps/linenoise/ diredeps/lua/src)
install(TARGETS redis-server
        RUNTIME DESTINATION bin
        )
#set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -c")
  • /src/modules/CMakeLists.txt
1
2
3
4
5
6
7
8
9
10
scss复制代码cmake_minimum_required(VERSION 3.9)
set(CMAKE_BUILD_TYPE "Debug")
add_library(helloworld SHARED helloworld.c)
set_target_properties(helloworld PROPERTIES PREFIX "" SUFFIX ".so")
add_library(hellotype SHARED hellotype.c)
set_target_properties(hellotype PROPERTIES PREFIX "" SUFFIX ".so")
add_library(helloblock SHARED helloblock.c)
set_target_properties(helloblock PROPERTIES PREFIX "" SUFFIX ".so")
add_library(testmodule SHARED testmodule.c)
set_target_properties(testmodule PROPERTIES PREFIX "" SUFFIX ".so")
  • ./deps/CMakeLists.txt
1
2
3
scss复制代码add_subdirectory(hiredis)
add_subdirectory(linenoise)
add_subdirectory(lua)
  • ./deps/linenoise/CMakeLists.txt
1
scss复制代码add_library(linenoise linenoise.c)
  • ./deps/lua/CMakeLists.txt
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
bash复制代码set(LUA_SRC
        src/lauxlib.c
        src/liolib.c
        src/lopcodes.c
        src/lstate.c
        src/lobject.c
        src/print.c
        src/lmathlib.c
        src/loadlib.c
        src/lvm.c
        src/lfunc.c
        src/lstrlib.c
        src/lua.c
        src/linit.c
        src/lstring.c
        src/lundump.c
        src/luac.c
        src/ltable.c
        src/ldump.c
        src/loslib.c
        src/lgc.c
        src/lzio.c
        src/ldblib.c
        src/strbuf.c
        src/lmem.c
        src/lcode.c
        src/ltablib.c
        src/lua_struct.c
        src/lapi.c
        src/lbaselib.c
        src/lua_cmsgpack.c
        src/ldebug.c
        src/lparser.c
        src/lua_cjson.c
        src/fpconv.c
        src/lua_bit.c
        src/llex.c
        src/ltm.c
        src/ldo.c
        )
add_library(lua STATIC ${LUA_SRC})

编辑 ./src/ae_kqueue.c, 增加三个头文件，其中 unistd.h 是为了解决在本机上编译是有一个 c99 无 close 函数的报错

1
2
3
arduino复制代码#include "ae.h"
#include "zmalloc.h"
#include <unistd.h> // for close

Debug

此时应该有Debug选项了, 如果没有, 可以重新Relaod CMakelist.txt

图片名称
图片名称

参考
github.com/LHRchina/re…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%