发表于

这是我参与11月更文挑战的第5天，活动详情查看：2021最后一次更文挑战

题目

将两个升序链表合并为一个新的 升序 链表并返回。新链表是通过拼接给定的两个链表的所有节点组成的。

image.png

  • 示例 1：

输入：l1 = [1,2,4], l2 = [1,3,4]
输出：[1,1,2,3,4,4]

  • 示例 2：

输入：l1 = [], l2 = []
输出：[]

  • 示例 3：

输入：l1 = [], l2 = [0]
输出：[0]

提示：

两个链表的节点数目范围是 [0, 50]
-100 <= Node.val <= 100
l1 和 l2 均按 非递减顺序 排列

解题：

先了解什么是升序链表:

链表（Linked list）是一种常见的基础数据结构，是一种线性表

升序链表中，数据是按照关键值有序排列的，并且是从小到大排序

什么是递归算法？

递归算法（recursion algorithm）在计算机科学中是指一种通过重复将问题分解为同类的子问题而解决问题的方法。

阅读递归函数最容易的方法不是纠缠于它的执行过程，而是相信递归函数会顺利完成它的任务。而我们写递归函数要考虑的是每个步骤正确无误，限制条件设置正确，并且每次调用之后更接近限制条件，递归函数总是能正确的完成任务。

考虑边界值：

题目要求新链表是通过拼接给定的两个链表的所有节点组成的，那么有以下的情况

  • 其中一个为空链表，那么新链表则等于另一个链表
  • 两个都是空链表，这样拼接组成的新链表也是空链表
  • 两个链表都不为空：
    从两个链表节点的值进行判断，把节点值小的链表指向的下一节点传下去，而另外的链表则把当前节点传下去继续进行节点值的比较，知道递归调用中某一个链表参数为空，则退出递归调用，向上逐层返回链表结果

实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
java复制代码class Solution {
    public ListNode mergeTwoLists(ListNode list1, ListNode list2) {
        // 其中一个为空链表，那么新链表则等于另一个链表
        if(list1 == null){
            return list2;
        } else if(list2 == null){
            return list1;
        } else if(list1.val < list2.val){
            // 递归调用
            list1.next = mergeTwoLists(list1.next, list2);
            return list1;
        } else {
            // 递归调用
            list2.next = mergeTwoLists(list1, list2.next);
            return list2;
        }
    }
}

本文转载自: 掘金

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

发表于

「这是我参与11月更文挑战的第20天，活动详情查看：2021最后一次更文挑战」。

本文介绍了树、森林和二叉树之间的转换策略，并且附有相关图片演示。最后还介绍了树与森林的通用遍历方式。对二叉树不了解的，建议先看这篇文章：二叉树的入门以及Java实现案例详解。

1 树转换为二叉树

对树采用孩子兄弟表示法即可，关于孩子兄弟表示法，可以看这篇文章：树结构的入门以及Java通用实现方式，其中的实现方法中有介绍。

树转换为二叉树的具体步骤：

  1. 加线。在所有兄弟结点之间加一条连线。
  2. 去线。对树中每个结点，只保留它与第一个孩子结点的连线，删除它与其他孩子结点之间的连线。
  3. 层次调整。以树的根结点为轴心，将整棵树顺时针旋转一定的角度，使之结构层次分明。注意第一个孩子是二叉树结点的左孩子，兄弟转换过来的孩子是结点的右孩子。

1.1 案例

如下一颗普通的树：

在这里插入图片描述

添加连线。在所有兄弟结点之间加一条连线。加线使用红色线表示：

在这里插入图片描述

删除连线。对树中每个结点，只保留它与第一个孩子结点的连线，删除它与其他孩子结点之间的连线。删除线使用虚线表示：

在这里插入图片描述

调整结构：

在这里插入图片描述

2 森林转换为二叉树

森林是由若干棵树组成的，所以完全可以理解为，森林中的每一棵树都是兄弟，可以按照兄弟的处理办法来操作。步骤如下：

  1. 把每个树转换为二叉树。
  2. 第一棵二叉树不动，从第二棵二叉树开始，依次把后一棵二叉树的根结点作为前一棵二叉树的根结点的右孩子，用线连接起来。当所有的二叉树连接起来后就得到了由森林转换来的二叉树。

2.1 案例

在这里插入图片描述

3 二叉树转换为树

二叉树转换为树是树转换为二叉树的逆过程，也就是反过来做而已。步骤如下：

  1. 加线。若某结点的左孩子结点存在，则将这个左孩子的右孩子结点、右孩子的右孩子结点、右孩子的右孩子的右孩子结点……都作为此结点的孩子。将该结点与这些右孩子结点用线连接起来。
  2. 去线。删除原二叉树中所有结点与其右孩子结点的连线。
  3. 层次调整。使之结构层次分明。

3.1 案例

如下二叉树：

在这里插入图片描述

加线。若某结点的左孩子结点存在，则将这个左孩子的右孩子结点、右孩子的右孩子结点、右孩子的右孩子的右孩子结点……都作为此结点的孩子。将该结点与这些右孩子结点用线连接起来。加线使用红色线表示。

在这里插入图片描述

去线。删除原二叉树中所有结点与其右孩子结点的连线。删除线使用虚线表示：

在这里插入图片描述

调整结构：

在这里插入图片描述

4 二叉树转换为森林

判断一棵二叉树能够转换成一棵树还是森林，标准很简单，那就是只要看这棵二叉树的根结点有没有右孩子，有就是森林，没有就是一棵树。那么如果是转换成森林，步骤如下：

  1. 从根结点开始，若右孩子存在，则把与右孩子结点的连线删除，再查看分离后的二叉树，若右孩子存在，则连线删除……，直到所有右孩子连线都删除为止，得到分离的二叉树。
  2. 再将每棵分离后的二叉树转换为树即可。

4.1 案例

在这里插入图片描述

5 树与森林的遍历

5.1 树的遍历

树的遍历分为两种方式：

  1. 一种是先根遍历树，即先访问树的根结点，然后依次先根遍历根的每棵子树，并且从左至右遍历树的子节点。
  2. 另一种是后根遍历，即先依次后根遍历每棵子树节点，然后再访问根结点。

5.1.1 案例

在这里插入图片描述

上面的树，它的先根遍历序列为radijkebfcgh，后根遍历序列为ijkdeafbghcr。先根遍历、后根遍历的顺序实际上和该树转换为二叉树后的先序遍历、中序遍历是一致的。

关于二叉树的遍历，可以看这篇文章：二叉树的4种遍历方式详解以及Java代码的完整演示。

5.2 森林的遍历

森林的遍历也分为两种方式：

  1. 前序遍历：先访问森林中第一棵树的根结点，然后再依次先根遍历根的每棵子树，再依次用同样方式遍历除去第一棵树的剩余树构成的森林。
  2. 后序遍历：是先访问森林中第一棵树，后根遍历的方式遍历每棵子树，然后再访问根结点，再依次同样方式遍历除去第一棵树的剩余树构成的森林。

森林的前序遍历和转换为二叉树后对二叉树的前序遍历结果相同，森林的后序遍历和转换为二叉树后对二叉树的中序遍历结果相同。

如果有什么不懂或者需要交流，可以留言。另外希望点赞、收藏、关注，我将不间断更新各种Java学习博客！

本文转载自: 掘金

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

发表于

MySQL支持的锁

从锁粒度上划分

1
2
3
scss复制代码表级锁
行级锁(InnoDB存储引擎)
页级锁(BDB存储引擎)

从锁操作上划分

从实现方式上划分

使用场景

修改表结构

1
复制代码修改数据库表结构会自动加表级锁（元数据锁）

行级锁升级表级锁

1
2
复制代码更新数据未使用索引
行级锁会上升为表级锁

更新数据使用索引会使用行级锁

select .... from update使用行级锁

MySQL锁分类

1
复制代码分为乐观锁和悲观锁

乐观锁

1
复制代码乐观锁是程序通过版本号或时间戳实现

悲观锁

表级锁

1
2
3
4
5
复制代码每次操作锁住整个表
锁定力度大
发生锁冲突的概率最高
并发度最低
应用在MyISAM、InnoDB、BDB等存储引擎中

1
2
3
scss复制代码表级锁又分为表锁（MySQL layer层加锁）
元数据锁（MySQL layer层加锁）
意向锁(InnodB存储引擎层加锁) 内部使用的锁
表锁
1
复制代码需要手动加锁

  • read lock
1
2
复制代码加读锁后还可以加读锁
不能加写锁
  • write lock
1
复制代码加写锁后不能加读锁也不能加写锁
元数据锁
1
2
复制代码自动加锁
元数据其实就是表结构

意向锁

行级锁

1
2
3
4
5
复制代码每次操作锁住一行数据
锁定粒度最小
发生锁冲突的概率最低
并发读最高
是由InnoDB存储引擎实现的

共享读锁(S)
1
2
3
4
csharp复制代码手动加锁
select .... lock in share mode
允许一个事务去读一行 
阻止其他事务获取相同数据集的排他锁
排他写锁(X)
1
2
3
4
复制代码自动加锁
允许获得排他写锁的事务更新数据
阻止其他事务取得相同数据集的共享读锁（不是普通读）
和排他写锁
  • DML(insert、update、delete)
  • select … from udpate

整体分类

表级锁使用

表读锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
bash复制代码事务1给mylock表添加读锁
事务1查询mylock表可以读取到数据
事务1不可以查询其他表数据

事务2普通查询mylock表 
没有加锁 
可以查到 
通过MVCC机制查询数据的历史版本

事务2更新mylock表id为2的数据 
需要先获取这条数据的行锁
才可以进行修改
但此时是获取不到的
因为事务1还未释放mylock表的读锁
所以事务2只能等待事务1释放mylock表的读锁
才能够获取到行锁

事务1释放了mylock表的读锁
那么则可以查询其他表的数据了

事务2也获取到id为2的这条数据的行锁
执行更新操作

表写锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
bash复制代码事务1给mylock添加表写锁
事务1可以查询mylock数据
事务1不可以查询其他表数据
事务1更新mylock表id为2的数据
可以执行

事务2查询mylock数据
查询阻塞 
因为事务1还未释放mylock表的写锁

事务1释放mylock表的写锁

事务2的查询得以继续执行
事务1页可以访问其他表数据了

元数据锁的使用

元数据读锁

1
2
3
4
5
6
7
8
9
复制代码事务1开启事务
事务1查询mylock表
此时会加一个MDL读锁

事务2修改mylock表结构
此时会被阻塞

事务1提交事务或回滚事务
事务2才得以修改完成

行级锁分类及使用

查询行级锁状态

1
sql复制代码show status like 'innodb_row_locks'

行级锁的使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
bash复制代码事务1 开始事务
事务1 查询mylock表id为1的数据
id列为索引列
加行读锁

事务2更新id为2的数据 
因未锁定该行 所以可以更新

事务2更新id为1的数据
该行的行读锁还未释放
此时修改被阻塞

事务1提交事务
事务2对id为1这条数据的更新才得以执行

"注"
使用索引加行锁
未锁定的行可以访问

行读锁升级为表锁

未使用索引的行级锁会升级为表锁

1
2
3
4
5
6
7
8
9
10
11
12
ini复制代码事务1开始事务
事务1查询mylock表 查询条件是name='c'
并手动给name='c'添加行读锁
但name列并未使用索引
所以行读锁就会升级为表级锁

事务2更新mylock表id为2的数据
就会被阻塞

事务1提交或回滚事务 表级锁就会被释放
事务2的更新操作就可以获取到行级锁 
然后执行更新操作

行写锁

主键索引产生记录锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
bash复制代码事务1开始事务
事务1查询mylock表id为1的数据 
并添加行写锁

事务2查询mylock表id为2的数据 
可以访问

事务2查询mylock表id为1的数据
可以访问 不加锁

事务2 查询mylock表id为1的数据 添加读锁
此时被阻塞

事务1提交 释放了id为1的行写锁
事务2 加读锁获取id为1的数据得以执行

行锁原理

主键加锁

1
2
3
bash复制代码id为主键索引
update t set name='x' where id =10;
加锁行为仅在id=10的主键索引记录上加X锁(记录锁)

唯一键加锁

1
2
3
4
5
bash复制代码id为唯一键索引
name为主键索引

先在唯一索引id上加id=10的x锁
再在id=10的主键索引记录上加x锁

非唯一键加锁

1
2
3
4
5
6
7
8
9
scss复制代码name是主键
id是索引（二级索引）

加锁行为：对满足id=10条件的记录和主键分别加x锁
然后在
(6,c)~(10,b)
(10,b)~(10,d)
（10,d)~(11,f)
间隙分别加Gap锁

1
2
复制代码有了间隙锁 
这些区间内不允许其他事务做插入操作

无索引加锁

1
2
3
4
5
6
7
8
bash复制代码name是逐渐
id没有索引
select * from t where id =10; 
会导致全表扫描

加锁行为：表里所有行和间隙均加x锁
由于InnoDB引擎行锁机制基于索引实现记录锁定
因为没有索引时会导致全表锁定

死锁

死锁现象

  • 表锁死锁
  • 行级锁死锁
  • 共享锁转换为排他锁

表级锁死锁

1
2
3
4
5
6
7
8
9
css复制代码用户A先访问表A 对表A加了锁
然后再访问表B
用户B先访问表B 对表B加了锁
然后再访问表A
因表B被加了锁
所以用户A需等着用户B释放了表B的锁才可以对表B加锁
因表A被加了锁
所以用户B需等着用户A释放了表A的锁才可以对表A加锁
所以2者互相等待 从而死锁

解决方案

  • 调整程序的逻辑
1
css复制代码把表A和表B当成同一个资源

1
2
css复制代码用户A访问完表A和表B之后
用户B再来访问表A和表B
  • 尽量避免同时锁定2个资源

行级锁死锁

产生原因1

1
2
3
4
5
rust复制代码在事务中执行了一条不满足for update的操作
则执行全表扫描
把行级锁上升为表级锁
多个这样的事务执行后
就很容易产生死锁和阻塞

解决方案

1
2
sql复制代码SQL语句中不要使用太复杂的关联表的查询 
优化索引

产生原因2

1
2
3
4
5
6
7
8
9
10
11
12
复制代码表中的2个数据id1和id2
事务1先访问id1 对id1加行锁
再访问id2
事务2先访问id2 对id2加行锁
再访问id1
事务1访问id2等待事务2释放id2的行锁
事务2访问id1等待事务1释放id1的行锁
所以事务1和事务2互相等待 阻塞
从而产生死锁

类似于2个表死锁
这个是表中的2个记录行级死锁

解决方案

  • 同一个事务中 尽可能做到一次性锁定所有资源
  • 按照id对资源排序 然后按顺序进行处理
  • 采用MVCC机制处理 普通读 不会使用锁

共享锁转排他锁

1
2
3
4
5
6
7
8
9
10
css复制代码事务A查询一条记录 加共享读锁
事务A更新这条记录
事务B也更新这条记录 需要排他锁
事务B需等待事务A释放了共享锁
才可以获得排他锁进行更新
所以事务B进入了排队等待
事务A也需要排他锁进行更新操作
但是无法授予该 锁请求
因为事务B已经有了一个排他锁请求
并且等待事务A释放其共享锁

解决方案

  • 避免引发对同一条记录的反复操作
  • 使用乐观锁进行控制

本文转载自: 掘金

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

发表于

go run main.go 一个 Go 程序就启动了。然而这背后操作系统如何执行到 Go 代码的，Go 为了运行用户 main 函数，又做了什么？

一 编译

  • go build main.go

我们写的 go 代码都是编译成可执行文件去机器上直接执行的，在 linux 平台上是 ELF 格式的可执行文件，linux 能直接执行这个文件。

  • 编译器：将 go 代码生成 .s 汇编代码，go 中使用的是 plan9 汇编
  • 汇编起：将汇编代码转成机器代码，即目标程序 .o 文件
  • 链接器：将多个 .o 文件合并链接得到最终可执行文件

go程序汇编代码.o目标程序可执行文件写代码编译器汇编器链接器结束
二 操作系统加载

  • ./main

经上述几个步骤生成可执行文件后，二进制文件在被操作系统加载起来运行时会经过如下几个阶段：

  1. 从磁盘上把可执行程序读入内存；
  2. 创建进程和主线程；
  3. 为主线程分配栈空间；
  4. 把由用户在命令行输入的参数拷贝到主线程的栈；
  5. 把主线程放入操作系统的运行队列等待被调度执起来运行；

START_THREAD(elf_ex, regs, elf_entry, bprm->p) 启动线程传入了 elf_entry 参数，这是程序的入口地址。

这个 elf_entry 被写在 elf 可执行文件的 header 中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
bash复制代码$ readelf -h main
ELF Header:
  Magic:   7f 45 4c 46 02 01 01 00 00 00 00 00 00 00 00 00
  Class:                             ELF64
  Data:                              2's complement, little endian
  Version:                           1 (current)
  OS/ABI:                            UNIX - System V
  ABI Version:                       0
  Type:                              EXEC (Executable file)
  Machine:                           Advanced Micro Devices X86-64
  Version:                           0x1
  Entry point address:               0x45d430
  Start of program headers:          64 (bytes into file)
  Start of section headers:          456 (bytes into file)
  Flags:                             0x0
  Size of this header:               64 (bytes)
  Size of program headers:           56 (bytes)
  Number of program headers:         7
  Size of section headers:           64 (bytes)
  Number of section headers:         25
  Section header string table index: 3

并且通过反编译 可执行文件，可以看到这个地址对应的就是 _rt0_amd64_linux。

1
2
3
4
shell复制代码$ objdump ./main -D > tmp
$ grep tmp '45d430'
000000000045d430 <_rt0_amd64_linux>:
  45d430:	e9 2b c4 ff ff       	jmpq   459860 <_rt0_amd64>

从此进入了 Go 程序的启动过程

Go 程序启动

Go 程序启动位置, 把栈上的入口参数存到寄存器中，接下来跳转到 rt0_go 启动函数

1
2
3
4
scss复制代码TEXT _rt0_amd64(SB),NOSPLIT,$-8
	MOVQ	0(SP), DI	// argc
	LEAQ	8(SP), SI	// argv
	JMP	runtime·rt0_go(SB)

rt0_go 代码比较长，可分为两个部分，第一部分是系统参数获取和运行时检查。第二部分是 go 程序启动的核心，这里只详细介绍第二部分，总体启动流程如下

rt0_gogo runtime 核心schedinitnewprocmstartruntime.main入口参数复制tls 线程本地存储初始化,关联当前线程,m0与g0check:运行时类型检查args:argc,argv,内存物理页系统参数的获取osinit:cpu核心获取schedinit:调度器初始化newproc:创建新的 goroutinemstart:开始调度栈的初始化堆内存分配的初始化当前G系统线程初始化垃圾回收器参数初始化通过cpu核心数和gomaxprocs创建P创建新的协程,入口地址 runtime.main放到P的队列中,等待调度初始化M,开始调度调度获取到newproc创建的G, 入口方法:runtime.main新建sysmon监视线程执行所有runtime init方法启动 gc 收集器执行所有用户包依赖的init方法执行main.main
go runtime 核心：

  1. schedinit：进行各种运行时组件初始化工作，这包括我们的调度器与内存分配器、回收器的初始化
  2. newproc：负责根据主 goroutine（即 main）入口地址创建可被运行时调度的执行单元，这里的main还不是用户的main函数，是 runtime.main
  3. mstart：开始启动调度器的调度循环, 执行队列中 入口方法是 runtime.main 的 G
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
c复制代码TEXT runtime·rt0_go(SB),NOSPLIT,$0
	(...)
	// 调度器初始化
	CALL	runtime·schedinit(SB)

	// 创建一个新的 goroutine 来启动程序
	MOVQ	$runtime·mainPC(SB), AX
	PUSHQ	AX
	PUSHQ	$0			// 参数大小
	CALL	runtime·newproc(SB)
	POPQ	AX
	POPQ	AX

	// 启动这个 M，mstart 应该永不返回
	CALL	runtime·mstart(SB)
	(...)
	RET

shedinit包括了所有核心组件的初始化工作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
go复制代码// src/runtime/proc.go
func schedinit() {
	_g_ := getg()
	(...)

	// 栈、内存分配器、调度器相关初始化
	sched.maxmcount = 10000	// 限制最大系统线程数量
	stackinit()			// 初始化执行栈
	mallocinit()		// 初始化内存分配器
	mcommoninit(_g_.m)	// 初始化当前系统线程
	(...)

	gcinit()	// 垃圾回收器初始化
	(...)

	// 创建 P
	// 通过 CPU 核心数和 GOMAXPROCS 环境变量确定 P 的数量
	procs := ncpu
	if n, ok := atoi32(gogetenv("GOMAXPROCS")); ok && n > 0 {
		procs = n
	}
	procresize(procs)
	(...)
}

执行 runtime.main, 主要进行了

  1. 启动系统后台监控sysmon 线程
  2. 执行 runtime 包内 init
  3. 启动gc
  4. 用户包依赖 init 的执行
  5. 执行用户main.mian 方法
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
go复制代码// The main goroutine.
func main() {
    g := getg()

    ...
    // 执行栈最大限制：1GB（64位系统）或者 250MB（32位系统）
    if sys.PtrSize == 8 {
        maxstacksize = 1000000000
    } else {
        maxstacksize = 250000000
    }
    ...

    // 启动系统后台监控（定期垃圾回收、抢占调度等等）
    systemstack(func() {
        newm(sysmon, nil)
    })

    ...
    // 让goroute独占当前线程， 
    // runtime.lockOSThread的用法详见http://xiaorui.cc/archives/5320
    lockOSThread()

    ...
    // runtime包内部的init函数执行
    runtime_init() // must be before defer

    // Defer unlock so that runtime.Goexit during init does the unlock too.
    needUnlock := true
    defer func() {
        if needUnlock {
                unlockOSThread()
        }
    }()
    // 启动GC
    gcenable()

    ...
    // 用户包的init执行
    main_init()
    ...

    needUnlock = false
    unlockOSThread()

    ...
    // 执行用户的main主函数
    main_main()
    
    ...
    // 退出
    exit(0)
    for {
        var x *int32
        *x = 0
    }
}

总结

启动一个 Go 程序时，首先要经过操作系统的加载，通过可执行文件中 Entry point address 记录的地址，找到 go 程序启动入口: _rt0_amd64 -> rt0_go。rt0_go 中 先进行了 go 程序的 runtime 的初始化，其中包括:调度器，栈，堆内存空间初始化，垃圾回收器的初始化，最后最后通过newproc和mstart调度执行runtime.main,完成一系列初始化过程，再然后才是执行用户的主函数。

参考

  1. www.bookstack.cn/read/qcrao-…
  2. eddycjy.com/posts/go/go…
  3. loulan.me/post/golang…
  4. juejin.cn/post/694250…

本文转载自: 掘金

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

发表于

「这是我参与11月更文挑战的第11天，活动详情查看：2021最后一次更文挑战」

描述

统计一个数字在排序数组中出现的次数。

  • 示例 1:
1
2
ini复制代码输入: nums = [5,7,7,8,8,10], target = 8
输出: 2
  • 示例 2:
1
2
ini复制代码输入: nums = [5,7,7,8,8,10], target = 6
输出: 0
  • 提示
1
2
3
4
diff复制代码0 <= nums.length <= 105
-109 <= nums[i] <= 109
nums 是一个非递减数组
-109 <= target <= 109
  • 进阶：
  • 如果给定的数组中含有负数会发生什么？问题会产生何种变化？如果允许负数出现，需要向题目中添加哪些限制条件？

解析

由题目可知，在排序数组中，寻找target，是否在数组存在，已经存在的次数；如果没有存在就返回0,否则就返回存在的次数。

直观的思路肯定是从前往后遍历一遍。用两个变量记录第一次和最后一次遇见 \textit{target}target 的下标，但这个方法的时间复杂度为 O(n)O(n)，没有利用到数组升序排列的条件。

由于数组已经排序，因此整个数组是单调递增的，我们可以利用二分法来加速查找的过程。

如果存在target在数组中存在，那么他在数组中可以用下标给标示出来，那么我们可以找出数组中target出现的左边界left和有边界right,即返回的次数就是right-left-1。

示例

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
ini复制代码class Solution {
    public int search(int[] nums, int target) {
       //获取数组长度
       int length=nums.length;
       //因为这个数组里面的数字是排序好的，
       //所以通过二分法来确定target对应在数组中出现的区域，区域内都是target值
       int i=0;
       int j=length-1;
        //先确定目标区域的右边界
       while(i<=j){
           int m=(i+j)/2;
           //if存在说明target所在范围在二分法的右边
           if(nums[m]<=target){
               i=m+1;
           }else{
               j=m-1;
           }
       }
​
       int right=i;
       if(j>0&&nums[j]!=target){
           return 0;
       }
       i=0;j=length-1;
       //确定目标的左边界
       while(i<=j){
           int m=(i+j)/2;
           //target的左边界必然小于target的值
           if(nums[m]<target){
               i=m+1;
           }else{
               j=m-1;
           }
       }
       int left=j;
        //右区间-左区间-1 就是含的个数
       return right-left-1;
    }
}
​

运行结果：

执行结果：通过

执行用时：0 ms, 在所有 Java 提交中击败了100.00%的用户

内存消耗：41.1 MB, 在所有 Java 提交中击败了81.55%的用户

本文转载自: 掘金

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

发表于

前言

前面讲了基于短连接的saltstack主要功能实现，今天我来讲讲基于长连接方式的saltstack实现。

长连接

长连接和短连接的区别

HTTP长连接、短连接究竟是什么？

使用的长连接实现模块

websocket-client
gevent-websocket

这里没有使用websockets这个库来做长连接的客户端和服务端，主要考虑的是上面2个库来做客户端和服务端的代码更容易理解

由于以上2个模块都属于第三方库，所以需要使用如下命令先行安装

1
shell复制代码pip install websocket-client gevent-websocket

具体代码实现

长连接服务端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
python复制代码from geventwebsocket import WebSocketServer, WebSocketApplication, Resource
from collections import OrderedDict


class EchoApplication(WebSocketApplication):

    def on_open(self):
        print("Connection opened")

    def on_message(self, message):
        print(message)
        self.ws.send('recieve message')

    def on_close(self, reason):
        print(reason)


if __name__ == '__main__':
    WebSocketServer(
        ('0.0.0.0', 8761),
        Resource(OrderedDict([('/ws', EchoApplication)]))
    ).serve_forever()

以上代码很容易理解，当程序启动时监听8761端口，当有客户端连入时，on_open方法将会被执行，收到消息时，on_message将会被执行，当长连接关闭时，on_close方法将会被执行。

长连接客户端

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
python复制代码import websocket


def on_message(ws, message):
    print(message)


def on_error(ws, error):
    print(ws)
    print(error)


def on_open(ws):
    ws.send("agent connect to master")


def on_close(ws, close_status_code, close_msg):
    print("### closed ###")
    print(close_status_code, close_msg)


def keep_alive():
    websocket.enableTrace(False)
    ws = websocket.WebSocketApp(
        "ws://127.0.0.1:8761/ws",
        on_message=on_message,
        on_open=on_open,
        on_error=on_error,
        on_close=on_close,
    )

    ws.run_forever(suppress_origin=True, skip_utf8_validation=True)


if __name__ == "__main__":
    keep_alive()

可以看到长连接客户端和服务端代码非常相似，这也是选择这2个模块开发的原因。

客户端启动时连接服务端的接口，需要特别注意的是websocket的地址是以ws开头（SSL的长连接以wss开头）

此时，我们运行长连接服务端和客户端

1
2
3
4
5
6
python复制代码root@demo:/data# python3 master_socket.py
Connection opened
agent connect to master

root@demo:/data# python3 agent_socket.py
recieve message

如上，这就代表客户端和服务端长连接已经建立起来。

此时，我们需要考虑一个问题，假如我们把长连接服务端作为salt-master,客户端作为salt-minion,我们需要怎样做，才能通过salt-master发送指令给salt-minion去执行并且返回执行结果了？

我们可以看到不管是长连接服务端还是客户端都没有提供类似短连接的接口，也就不存在外部直接发个请求就可以返回数据回来

我们再看服务端，ws.send()方法即是发送消息给客户端，理论上来说只要我们能控制这个ws对象，就可以发送消息给客户端，那如何控制了？比如说初始化一个ws全局变量，然后on_open时，将on_open(ws)中的ws参数赋值给ws全局变量，然后我们外部只要操作这个ws就可以了？

但这个方法其实是行不通的，我们的salt-master是控制批量主机的，如此多的salt-minion通过长连接和master通信，ws这个全局变量到底属于哪个远程主机长连接的了？

基于MQ实现长连接master和agent交互的纽带

其实，我们可以新增一个消息提供者，用于外部操作salt-master

我们来看下代码实现：

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
python复制代码from geventwebsocket import WebSocketServer, WebSocketApplication, Resource
from collections import OrderedDict
import redis
from threading import Thread


class EchoApplication(WebSocketApplication):

    def subscribe_msg(self):
        redis_conn = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True) 
        ps = redis_conn.pubsub()
        ps.subscribe('ws')
        for item in ps.listen():		#监听状态：有消息发布了就拿过来
            if item['type'] == 'message':
                msg = item['data']
                print(msg)
                self.ws.send(msg)

    def on_open(self):
        Thread(target=self.subscribe_msg).start()
        print("Connection opened")

    def on_message(self, message):
        print(message)
        self.ws.send('recieve message')

    def on_close(self, reason):
        print(reason)


if __name__ == '__main__':
    WebSocketServer(
        ('0.0.0.0', 8761),
        Resource(OrderedDict([('/ws', EchoApplication)]))
    ).serve_forever()

这里我们使用Redis作为MQ，主要是安装简单、Python操作也简单，也可以使用其他MQ（RabbitMQ、Kafka）,或者使用saltstack使用的zeromq（这里没有看过saltstack的源代码实现，估摸着saltstack也是通过zeromq来让外部操作salt-master的）

以上代码很简单，我们在长连接服务端新加了个方法，用于操作外部Redis，监听外部Redis的发布订阅模块，如果ws这个key有消息生成，这边就会消费它，即通过ws.send()方法发送给客户端；然后我们在on_open方法中起一个线程去处理这个操作外部Redis的操作。此时，我们只要在外部操作这个Redis的pubsub中的ws key,新增消息，例如{'ip': xxx, 'cmd': 'df -h'},我们就可以根据IP发送给具体的salt-minion去执行具体的命令

长连接客户端如何执行服务端发过来的消息

其实我们前面讲了短连接的实现，就是为了长连接做铺垫，比如说执行命令，我们只要将短连接的执行命令代码稍作修改搬过来用即可

代码如下：

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
python复制代码import websocket

def exec_cmd(cmd):
    # 基于subprocess.Popen方法执行本地shell命令
    proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
    if proc:
        result = proc.stdout.read()
        return result
    else:
        return None

def on_message(ws, message):
    # 获取服务端发送过来的命令，然后执行返回给服务端
    result = exec_cmd(message.get('cmd'))
    ws.send(result)
    print(message)


def on_error(ws, error):
    print(ws)
    print(error)


def on_open(ws):
    ws.send("agent connect to master")


def on_close(ws, close_status_code, close_msg):
    print("### closed ###")
    print(close_status_code, close_msg)


def keep_alive():
    websocket.enableTrace(False)
    ws = websocket.WebSocketApp(
        "ws://127.0.0.1:8761/ws",
        on_message=on_message,
        on_open=on_open,
        on_error=on_error,
        on_close=on_close,
    )

    ws.run_forever(suppress_origin=True, skip_utf8_validation=True)


if __name__ == "__main__":
    keep_alive()

这里只是将执行命令的方法搬过来，上传下载文件原理类似，这里不再赘述

最后

近几篇文章我们通过SSH实现了ansible主要功能、通过短连接和长连接实现saltstack主要功能，下一篇文章我们来讲讲如何模拟saltstack和ansible的命令行来操作远程主机

相关系列文章同步发布于个人博客
Jackless

模拟 saltstack/ansible 系列一（序言）

模拟 saltstack/ansible 系列二（实现 ansible 主要功能）

模拟 saltstack/ansible 系列三（基于短连接实现 saltstack 主要功能）

模拟 saltstack/ansible 系列四（基于长连接实现 saltstack 主要功能）

本文转载自: 掘金

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

发表于

「这是我参与11月更文挑战的第28天，活动详情查看：2021最后一次更文挑战」

今天说一下翻转字符串里的单词

image.png

实现思路

今天这个题目还是比较简单的，我们可以直接使用PHP内置的函数操作字符串，返回翻转后的结果。

空格有一个或多个，出现在字符串首尾或单词中间。

  • 我们可以先去掉首尾的空格，然后再把单词中间的空格变成1个，再把每个单词放入数组中，最后翻转数组输出每个元素。
  • 我们也可以直接遍历字符串，记录下每个单词放入数组中，最后翻转数组输出每个元素。

上面这两个方法使用了PHP内置的函数或者遍历多次字符串。下面说一下只遍历一次字符串就可以得出结果的方法。

首先我们反向遍历整个字符串，不管两个单词中间有多少空格，只要当前位置为空格，并且当前位置的前一个位置不是空格，就说明到了一个单词的边缘，需要把这个单词提取出来。

由于只是翻转字符串单词的顺序，并不是要每个单词字母也翻转，所以需要先把遍历的字符串的每个字符存入一个变量中，待遍历到单词边缘时，把整个单词放入结果中。

  • 遇到非空格时，把当前字符放入一个变量中。
  • 遇到空格时，判断当前位置的前一个字符是否为空格，如果不是空格，说明在一个单词的边缘，则把临时存放的单词放入结果中。

上面是一个大概的思路，还有一些首尾空格的处理问题，下面跟着代码详细讲一下。

完整代码

image.png

第725行代码，获取字符串长度，这里假设真个字符串没有中文。

第726行代码，定义一个变量，翻转后的结果字符串。

第727行代码，定义一个变量，临时存放单个单词。

第728-746行代码，遍历整个字符串。

第729-736行代码，当前遍历的字符为非空格时，把当前字符放入临时存放单词的变量$currentWord中，由于是反向遍历字符串，所以需要每次把当前字符当作开头与$currentWord连接起来。

如果当前遍历位置为0，也就是字符串的第一个字符位置。也就是说字符串的第一个位置为非空格时，由于已经没有其他空格字符当作整个单词的边界去判断，所以需要把当前存放的单词与整个翻转的字符串$execWords连接。

第737-745行代码，当前字符为空格，并且上一个位置（由于反向遍历字符串，所以上一个位置是$i+1）字符为非空格时，是一个单词的边界位置，此时说明存放单词的变量中已经是一个完整的单词了。如果上一个位置也是空格则表示是连续的空格。

为什么有这个判断呢？

1
php复制代码$i < $len - 1

因为当第一个位置的字符就是空格时，此时没有上一个位置，所以需要从第二个位置开始判断。

符合上面的条件，则把当前存放的单词与整个翻转的字符串$execWords连接，并把存放的单词变量$currentWord清空，以便记录下一个单词。

第748行代码，返回翻转后的字符串。

代码中732-734和739-741行代码没有具体说明，这个判断是处理首尾空格问题的，当翻转后字符串需要连接一个新单词时，不管是通过

1
php复制代码$execWords .= ' ' . $currentWord;

还是通过

1
php复制代码$execWords .= $currentWord . ' ';

都会造成首个单词前面或者最后一个单词后面出现一个空格。通过这个判断就可以避免这种清空。

另外我们也可以去掉这两个判断，使用下面的方式，得到完整的翻转后的字符串再处理。

1
php复制代码$execWords = trim($execWords);

只不过是使用了PHP的内置函数。

下面是在leetCode上执行的结果，感觉还不错。

Snipaste_2021-11-28_19-47-13.png

本文转载自: 掘金

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

发表于

前言

前面讲了基于SSH的ansible实现，今天我来讲讲基于短连接方式的saltstack实现。

为什么要讲基于短连接的saltstack实现呢？

前面不是说saltstack是基于长连接的吗？原因就在于当你掌握了短连接的实现后，长连接就水到渠成了，你也更能理解到短连接和长连接实现的一个区别

Agent设计

不管是短连接实现还是长连接实现，都需要在远程主机起一个服务来承担Agent角色，就如同saltstack的salt-minion一样。

下面我们就基于flask来简单实现一个短连接形式的agent（考虑性能的话可以使用sanic,这里用flask的原因是本人团队内部培训时，我主要培训的是flask）

由于flask属于第三方库，所以需要使用如下命令先行安装

1
shell复制代码pip install flask

如果想做成和salt-minion二进制模式启动的话，我们还需要安装一个额外的模块pyinstaller

1
shell复制代码pip install pyinstaller

以下是flask作为一个服务端基本的代码: app.py

1
2
3
4
5
6
7
8
9
10
python复制代码from flask import Flask

app = Flask(__name__)

@app.route("/")
def hello_world():
    return "Hello, World!"
    
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, debug=True)

当我们执行这个代码时，程序将会监听8080端口，请求http://127.0.0.1:8080将会返回Hello, World!

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
bash复制代码$ python3 app.py

 * Serving Flask app 'app' (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on all addresses.
   WARNING: This is a development server. Do not use it in a production deployment.
 * Running on http://127.0.0.1:8080/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 142-934-530

$ http http://127.0.0.1:8080

HTTP/1.0 200 OK
Content-Length: 13
Content-Type: text/html; charset=utf-8
Date: Wed, 24 Nov 2021 14:46:12 GMT
Server: Werkzeug/2.0.2 Python/3.10.0

Hello, World!

此时，我们可以通过pyinstaller将该程序编译成一个二进制程序

1
bash复制代码pyinstaller -F app.py

saltstack功能分析与代码实现

连接远程主机

基于短连接实现的agent，连接远程主机功能只需要判断接口是否能通即可，如上我们请求http://127.0.0.1:8080时返回200状态码，即可认为成功连接远程主机。

test.ping

ping功能的实现，我们可以写个ping接口，然后返回pong来实现，或者根据返回状态码是否为200来判断

我们把上边的app.py修改一下，加上ping接口

1
2
3
4
5
6
7
8
9
10
python复制代码from flask import Flask

app = Flask(__name__)

@app.route("/ping", methods=["GET"])
def ping():
    return "pong"
    
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, debug=True)

此时我们请求http://127.0.0.1:8080/ping则会返回pong,此时即可认为远程主机在线

1
2
3
4
5
6
7
8
9
python复制代码$ http http://127.0.0.1:8080/ping

HTTP/1.0 200 OK
Content-Length: 4
Content-Type: text/html; charset=utf-8
Date: Wed, 24 Nov 2021 15:04:58 GMT
Server: Werkzeug/2.0.2 Python/3.10.0

pong

cmd.run

执行命令我们可以再写一个接收POST请求的接口，接收一个参数，用于命令的传入，然后根据传入的命令在本地主机执行命令，获取命令返回的结果后响应给客户端。

我们把上边的app.py增加一个cmd接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
python复制代码from flask import Flask, request
import subprocess

app = Flask(__name__)

@app.route("/ping", methods=["GET"])
def ping():
    return "pong"
    
@app.route("/cmd", methods=["post"])
def cmd():
    body = request.json
    cmd = body.get("cmd")
    # 基于subprocess.Popen方法执行本地shell命令
    proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
    if proc:
        result = proc.stdout.read()
        return result
    else:
        return None
    
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, debug=True)

此时我们请求http://127.0.0.1:8080/cmd,传入命令，则会返回命令结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
python复制代码$ http http://127.0.0.1:8080/cmd cmd='df -h'

HTTP/1.0 200 OK
Content-Length: 590
Content-Type: text/html; charset=utf-8
Date: Wed, 24 Nov 2021 15:25:24 GMT
Server: Werkzeug/2.0.2 Python/3.8.10

Filesystem      Size  Used Avail Use% Mounted on
udev            957M     0  957M   0% /dev
tmpfs           198M  1.1M  197M   1% /run
/dev/sda1        20G  1.5G   18G   8% /
tmpfs           990M     0  990M   0% /dev/shm
tmpfs           5.0M     0  5.0M   0% /run/lock
tmpfs           990M     0  990M   0% /sys/fs/cgroup
/dev/sda15       98M  290K   98M   1% /boot/efi
/dev/loop0       61M   61M     0 100% /snap/lxd/21843
/dev/loop2       29M   29M     0 100% /snap/snapd/13643
/dev/loop1       58M   58M     0 100% /snap/core20/1244
tmpfs           198M     0  198M   0% /run/user/1000

上传文件

执行命令我们可以再写一个接收文件的接口，将客户端上传过来的文件保存到本地，可以接收一个目标目录的参数，用做存放文件的目录

我们把上边的app.py增加一个upload_file接口

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
python复制代码from flask import Flask, request
import subprocess

app = Flask(__name__)

@app.route("/ping", methods=["GET"])
def ping():
    return "pong"
    
@app.route("/cmd", methods=["post"])
def cmd():
    body = request.json
    cmd = body.get("cmd")
    # 基于subprocess.Popen方法执行本地shell命令
    proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
    if proc:
        result = proc.stdout.read()
        return result
    else:
        return None

@app.route("/upload_file", methods=["post"])
def upload_file():
    # 从请求中获取文件，如果上传的不是以file为key的文件，则认为未上传文件
    if "file" not in request.files:
        return "not find file in request.files"
    file = request.files["file"]
    # 如果请求没有带dest参数，则dest默认为/opt
    dest = request.args.get("dest", "/opt")
    if file:
        file.save(dest)
        return "file uploaded successfully"
    return "upload error"
    
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, debug=True)

客户端代码如下：

1
2
3
4
5
6
7
8
9
python复制代码def upload_file(src, dest):
    resp = requests.post(
        "http://127.0.0.1:8080/upload_file?dest=" + dest,
        files={"file": open(src, "rb")},
    )
    if resp.status_code == 200:
        return resp.text
    else:
        return None

此时客户端请求http://127.0.0.1:8080/upload_file,传入参数，则上传文件成功

下载文件

下载文件我们可以写个接口，接收文件所在目录地址和文件名2个参数

我们把上边的app.py增加一个download_file接口

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
python复制代码from flask import Flask, request
import subprocess

app = Flask(__name__)

@app.route("/ping", methods=["GET"])
def ping():
    return "pong"
    
@app.route("/cmd", methods=["post"])
def cmd():
    body = request.json
    cmd = body.get("cmd")
    # 基于subprocess.Popen方法执行本地shell命令
    proc = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
    if proc:
        result = proc.stdout.read()
        return result
    else:
        return None

@app.route("/upload_file", methods=["post"])
def upload_file():
    # 从请求中获取文件，如果上传的不是以file为key的文件，则认为未上传文件
    if "file" not in request.files:
        return "not find file in request.files"
    file = request.files["file"]
    # 如果请求没有带dest参数，则dest默认为/opt
    dest = request.args.get("dest", "/opt")
    if file:
        file.save(dest)
        return "file uploaded successfully"
    return "upload error"
    
@app.route("/download_file", methods=["post"])
def download_file():
    body = request.json
    directory = body.get("directory")
    filename = body.get("filename")
    # send_from_directory用于发送本地的文件
    return send_from_directory(
        directory=directory, filename=filename, as_attachment=True
    )

    
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, debug=True)

客户端代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
python复制代码def download_file(src, dest):
    filename = src.split("/")[-1]
    directory = src.replace(filename, "")
    print(directory, filename)
    data = {"directory": directory, "filename": filename}
    resp = requests.post("http://127.0.0.1:8080/download_file", json=data)
    if resp.status_code == 200:
        with open(dest, "wb") as f:
            f.write(resp.content)
        return True
    else:
        return False

以上接口基本上实现saltstack常用功能。下一文章我们来讲讲长连接实现saltstack基本功能

相关系列文章同步发布于个人博客
Jackless

模拟 saltstack/ansible 系列一（序言）

模拟 saltstack/ansible 系列二（实现 ansible 主要功能）

模拟 saltstack/ansible 系列三（基于短连接实现 saltstack 主要功能）

模拟 saltstack/ansible 系列四（基于长连接实现 saltstack 主要功能）

本文转载自: 掘金

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

发表于

〇. 准备工作

  • 导包
1
2
3
4
5
xml复制代码<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.20.Final</version>
</dependency>

一. Server端

  • server从宏观上来看，会处理两方面的事件：客户端连接请求、业务相关的事件（包括server端接受client端的消息、client端断开连接等等）。
  • 针对这两类事件，netty分别采用了两个线程组来实现。当有连接请求过来的时候，使用BossGroup进行处理；当连接成功之后，将该channel传递给WorkerGroup，WorkerGroup管理该channel的请求。
1
2
java复制代码EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

其中，值得说明的是，NioEventLoopGroup构造方法传递的参数，表示该线程组里面的线程池个数。上述表示，bossGroup有一个线程池负责处理连接请求。如果不提供参数，就默认为CPU核心数*2。

  • 之后就可以初始化netty server的启动类ServerBootstrap了
  • 首先，将上面初始化的两个EventLoopGroup作为参数放入。
  • 设置通道类型为NioServerSocketChannel
  • 设置已连接队列的大小，这个参数与TCP建立连接有关。详细可看：blog.csdn.net/weixin_4473…
  • 将连接设置为keep-alive，避免短时间内重复多次TCP握手。详细可看：www.cnblogs.com/caoweixiong…
  • 设置childHandler（实际上处理事件的是处理器链，后面再详细描述）。childHandler就是处理WorkerGroup的事件。在这首先添加一个初始化器，在初始化器里面给处理链添加了一个自定义的真正处理类的类MessageHandler。
1
2
3
4
5
6
7
8
9
10
11
java复制代码ServerBootstrap bootstrap = new ServerBootstrap()
        .group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .option(ChannelOption.SO_BACKLOG, 128)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                socketChannel.pipeline().addLast(new MessageHandler());
            }
        })
        .childOption(ChannelOption.SO_KEEPALIVE, true);
  • 一般处理器都会实现ChannelInboundHandlerAdapter或SimpleChannelInboundHandler。SimpleChannelInboundHandler是ChannelInboundHandlerAdapter的子类，会帮忙释放直接内存，避免内存泄露，因此继承SimpleChannelInboundHandler会比较方便。
  • channelRead0方法就是真正处理消息的方法
  • 指定client端发送ByteBuf类型的数据，因此接收的时候直接获取即可。（也可以指定不同类型的数据，需要指定不同的编解码器）
  • ChannelHandlerContext是该通道处理器相关的上下文对象，携带了通道本身、pipline等上下文。详细内容之后再分析。在本次代码中，通过上下文获取到channel本身，再通过channel获取到client端的ip地址。
  • 最好重写exceptionCaught，及时捕获异常并且进行处理。
1
2
3
4
5
6
7
8
9
10
11
12
13
java复制代码public class MessageHandler extends SimpleChannelInboundHandler<ByteBuf> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf byteBuf) throws Exception {
        System.out.println("receive message from " + ctx.channel().remoteAddress() + ", message: " + byteBuf.toString(CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(cause.getMessage());
        ctx.close();
    }
}
  • 完成启动类的初始化之后，还没完成。因为server端没有正式启动。
  • bind方法表示服务在PORT端口正式启动，并且调用sync方法，表示同步阻塞等待启动完成。
1
java复制代码ChannelFuture channelFuture = bootstrap.bind(PORT).sync();
  • 获取到channelFuture对象之后，就可以同步阻塞地监听关闭事件了。
1
java复制代码channelFuture.channel().closeFuture().sync();
  • 为了在关闭服务的时候，及时将线程池关闭，需要在finally时，关闭所有线程池
1
2
java复制代码bossGroup.shutdownGracefully(); 
workerGroup.shutdownGracefully();
  • 至此，最简单的一个netty server已完成。
  • 完整代码如下
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
java复制代码public class HelloWorldServer {
    public static final int PORT = 9999;

    public static void main(String[] args) {
        // set one thread for listening on port
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // depend on the number of core
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new MessageHandler());
                        }
                    })
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            System.out.println("finish initialize the parameter for boostrap.");
            System.out.println("begin to start netty server");

            ChannelFuture channelFuture = bootstrap.bind(PORT).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            System.out.println("shutdown bossGroup and workerGroup gracefully.");
        }
    }
}

二、Client端

  • 对于client端来说，会比server端少一个处理连接请求的EventLoopGroup，只有一个处理基本业务的EventLoopGroup。
1
java复制代码EventLoopGroup eventExecutors = new NioEventLoopGroup();
  • client的启动类为Bootstrap
  • 设置线程组为eventExecutors
  • 设置通道类型为NioSocketChannel（注意，与server端的是不一样的）
  • 因为只有线程组，因此handler也只有一组，因此在handler方法里面放一个初始化器就好。在初始化器里面再添加事件的真正处理类。
1
2
3
4
5
6
7
8
9
java复制代码Bootstrap bootstrap = new Bootstrap()
        .channel(NioSocketChannel.class)
        .group(eventExecutors)
        .handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new SendMessageHandler());
            }
        });
  • 这个处理器重写了channelActive方法，表示当通道创建成功之后就运行的方法。该方法用于接收用户输入的消息，并通过通道发送给server。
  • 其中，writeAndFlush方法可以实现写入缓存并发送给server。Unpooled.copiedBuffer可以将Sring类型的字符串变成netty自定义的一个二进制缓存类ByteBuf。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码public class SendMessageHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.print("please input you message: ");
        Scanner scan = new Scanner(System.in);
        while (scan.hasNextLine()) {
            String line = scan.nextLine();
            ctx.channel().writeAndFlush(Unpooled.copiedBuffer(line, CharsetUtil.UTF_8));
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
  • 启动类通过连接指定的地址、端口号实现连接，并且调用sync()方法实现同步阻塞等待。
  • 并且同步阻塞地等待关闭通道事件。
1
2
java复制代码ChannelFuture channelFuture = bootstrap.connect(ADDRESS, PORT).sync();
channelFuture.channel().closeFuture().sync();
  • 最后，需要保证在关闭client端之后，优雅地关闭线程组
1
java复制代码eventExecutors.shutdownGracefully();
  • 至此，完成了最简单的netty client端实现。
  • 以下是完整代码：
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
java复制代码public class HelloWorldClient {
    public static final String ADDRESS = "localhost";
    public static final int PORT = 9999;

    public static void main(String[] args) {
        EventLoopGroup eventExecutors = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap()
                    .channel(NioSocketChannel.class)
                    .group(eventExecutors)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new SendMessageHandler());
                        }
                    });

            System.out.println("finish initialize netty client");
            ChannelFuture channelFuture = bootstrap.connect(ADDRESS, PORT).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            eventExecutors.shutdownGracefully();
            System.out.println("shutdown eventExecutors gracefully.");
        }
    }
}

本文转载自: 掘金

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

发表于

前言

本次主要分享是如何基于Python模拟实现ansible的主要功能

ansible功能分析与代码实现

连接远程主机

ansible连接远程主机是通过SSH实现的，因此我们可以通过Python的SSH模块来实现连接远程主机。

Python中SSH模块主要是三方的paramiko模块，这个模块也是ansible内部实现SSH所用的模块。

由于paramiko属于第三方库，所以需要使用如下命令先行安装

1
bash复制代码pip install paramiko

下面是使用paramiko进行SSH连接和SFTP连接的相关代码

获取一个SSH连接

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
python复制代码def get_ssh_conn(hostname, port, username, password):
    try:
        # 实例化SSHClient
        ssh_client = paramiko.SSHClient()

        # 当远程服务器没有本地主机的密钥时自动添加到本地，这样不用在建立连接的时候输入yes或no进行确认
        ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        # 连接SSH服务器，这里以账号密码的方式进行认证，也可以用key认证
        ssh_client.connect(
            hostname=hostname, port=port, username=username, password=password, timeout=5
        )
        return ssh_client
    except Exception as e:
        logger.info(e)
        return None

获取一个SFTP连接

1
2
3
4
5
6
7
8
9
10
python复制代码def get_sftp_conn(hostname, port, username, password):
    try:
        # paramiko sftp功能需要先初始化一个Transport，有点类似打开一个文件传输通道
        t = paramiko.Transport((hostname, port))
        t.banner_timeout = 10
        t.connect(username=username, password=password)
        sftp = paramiko.SFTPClient.from_transport(t)
        return sftp
    except Exception as e:
        return None

执行命令

paramiko既然能连接主机，当然也能执行命令

基于SSH连接执行命令

1
2
3
4
python复制代码def exec_ssh_command(ssh_client, cmd):
    # 这个类似Popen执行命令，返回三个byte类型值，需要解码
    stdin, stdout, stderr = ssh_client.exec_command(cmd, timeout=30)
    return stderr.read().decode(), stdout.read().decode()

上传下载文件

当我们初始化好后SFTP连接后，就可以对远程服务器进行文件上传下载了

基于SFTP连接进行文件上传下载

1
2
3
4
5
6
7
python复制代码# put上传文件
def upload_file(sftp, src, dest):
    sftp.put(src, dest)

# get下载文件
def download_file(sftp, src, dest):
    sftp.get(src, dest)

ansible的ping功能

其实我们可以通过paramiko是否能ssh连接远程主机作为ping功能实现的原理，当ssh能连接成功时，会返回一个sshclient，否则返回None，参考获取一个SSH连接的代码

相关系列文章同步发布于个人博客
Jackless

模拟 saltstack/ansible 系列一（序言）

模拟 saltstack/ansible 系列二（实现 ansible 主要功能）

模拟 saltstack/ansible 系列三（基于短连接实现 saltstack 主要功能）

模拟 saltstack/ansible 系列四（基于长连接实现 saltstack 主要功能）

本文转载自: 掘金

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

0%