简介

在ASP.NET Core中,速率限制中间件是用来控制客户端对Web API或MVC应用程序发出请求的速率,以防止服务器过载和提高安全性。

下面是
AddRateLimiter
的一些基本用法:

1. 注册服务


Startup.cs

Program.cs
中,需要注册
AddRateLimiter
服务。这可以通过以下代码完成:

builder.Services.AddRateLimiter(options =>
{
    // 配置速率限制选项
});

2. 添加速率限制策略

可以添加不同类型的速率限制策略, 包括固定窗口、滑动窗口、令牌桶和并发限制。

固定窗口限制器(Fixed Window Limiter)

固定窗口限制器使用固定的时间窗口来限制请求。当时间窗口到期后,会开始一个新的时间窗口,并重置请求限制。例如,可以设置一个策略,允许每个12秒的时间窗口内最多4个请求。

builder.Services.AddRateLimiter(options =>
{
    options.AddFixedWindowLimiter("fixed", opt =>
    {
        opt.Window = TimeSpan.FromMinutes(1); // 时间窗口
        opt.PermitLimit = 3; // 在时间窗口内允许的最大请求数
        opt.QueueProcessingOrder = QueueProcessingOrder.OldestFirst; // 请求处理顺序
        opt.QueueLimit = 2; // 队列中允许的最大请求数
    });
});
app.UseRateLimiter();

即固定时间请求的次数,超过次数就会限流,下一个窗口时间将次数重置

经过测试,多余的请求还是会等待

https://www.cnblogs.com/guoxiaotian/p/17834892.html

滑动窗口限制器(Sliding Window Limiter)

滑动窗口算法:

  • 与固定窗口限制器类似,但为每个窗口添加了段。 窗口在每个段间隔滑动一段。 段间隔的计算方式是:(窗口时间)/(每个窗口的段数)。
  • 将窗口的请求数限制为 permitLimit 个请求。
  • 每个时间窗口划分为一个窗口 n 个段。
  • 从倒退一个窗口的过期时间段(当前段之前的 n 个段)获取的请求会添加到当前的段。 我们将倒退一个窗口最近过期时间段称为“过期的段”。

请考虑下表,其中显示了一个滑动窗口限制器,该限制器的窗口为 30 秒、每个窗口有三个段,且请求数限制为 100 个:

  • 第一行和第一列显示时间段。
  • 第二行显示剩余的可用请求数。 其余请求数的计算方式为可用请求数减去处理的请求数和回收的请求数。
  • 每次的请求数沿着蓝色对角线移动。
  • 从时间 30 开始,从过期时间段获得的请求会再次添加到请求数限制中,如红色线条所示。

下表换了一种格式来显示上图中的数据。 “可用”列显示上一个段中可用的请求数(来自上一个行中的“结转”)。 第一行显示有 100 个可用请求,因为没有上一个段。

时间 可用 获取的请求数 从过期段回收的请求数 结存请求数
0 100 20 0 80
10 80 30 0 50
20 50 40 0 10
30 10 30 20 0
40 0 10 30 20
50 20 10 40 50
60 50 35 30 45
 services.AddRateLimiter(options =>
    {
        options.AddSlidingWindowLimiter("sliding", opt =>
        {
            opt.Window = TimeSpan.FromMinutes(1); // 总窗口时间为1分钟
            opt.SegmentsPerWindow = 6; // 将1分钟的窗口分为6个段,即每10秒一个段
            opt.PermitLimit = 10; // 整个窗口时间内允许的最大请求数
        });
    });

令牌桶限制器(Token Bucket Limiter)

令牌桶限制器维护一个滚动累积的使用预算,作为一个令牌的余额。它以一定的速率添加令牌,当服务请求发生时,服务尝试提取一个令牌(减少令牌计数)来满足请求。如果没有令牌,服务就达到了限制,响应被阻塞。

    services.AddRateLimiter(configureOptions =>
    {
        configureOptions.AddTokenBucketLimiter("token-bucket", options =>
        {
            options.TokenLimit = 100; // 桶的容量
            options.ReplenishmentPeriod = TimeSpan.FromSeconds(10); // 补充周期,即每10秒补充一次令牌
            options.TokensPerPeriod = 10; // 每个周期补充的令牌数
            options.AutoReplenishment = true; // 是否自动补充令牌
            options.QueueProcessingOrder = QueueProcessingOrder.OldestFirst; // 队列处理顺序
            options.QueueLimit = 10; // 请求队列长度限制
        });
    });

并发限制器(Concurrency Limiter)

并发限制器是最简单的速率限制形式。它不关注时间,只关注并发请求的数量。

    services.AddRateLimiter(options =>
    {
        options.AddConcurrencyLimiter("concurrency", options =>
        {
            options.PermitLimit = 1; // 最大并发请求数
            options.QueueProcessingOrder = QueueProcessingOrder.OldestFirst; // 队列处理顺序
            options.QueueLimit = 10; // 请求队列长度限制
        });
    });

3. 使用中间件


Configure
方法或
Program.cs
中,需要使用
UseRateLimiter
中间件:

app.UseRateLimiter();

4. 应用速率限制策略

可以全局应用速率限制策略,或者将其应用于特定的控制器或动作:

全局配置

app.MapControllers().RequireRateLimiting("fixed");

应用于特定的控制器

[EnableRateLimiting("fixed")]
public class RateLimitTestController : ControllerBase
{
    // 控制器动作
}

应用于特定的动作

[EnableRateLimiting("fixed")]
public async Task<IActionResult> Get()
{
    // 动作逻辑
}

5. 禁用速率限制

也可以选择禁用速率限制,无论是在控制器级别还是特定动作级别:

禁用控制器级别的速率限制

[DisableRateLimiting]
public class RateLimitTestController : ControllerBase
{
    // 控制器动作
}

禁用特定动作的速率限制

[DisableRateLimiting]
public async Task<IActionResult> Get()
{
    // 动作逻辑
}

自定义响应

当客户端超出速率限制时,可以自定义响应。例如,可以设置
OnRejected
回调来自定义响应:

options.OnRejected = (context, token) =>
{
    context.HttpContext.Response.StatusCode = StatusCodes.Status429TooManyRequests;
    context.HttpContext.Response.Headers["Retry-After"] = "60"; // 建议60秒后重试
     context.HttpContext.Response.StatusCode = StatusCodes.Status429TooManyRequests;
 context.HttpContext.Response.WriteAsync("Too many requests. Please try again later.", cancellationToken: token);
                   
    return Task.CompletedTask;
};

总结

在ASP.NET Core应用程序中实现有效的速率限制策略,以保护的API免受滥用和过载。欢迎关注我的公众号:Net分享

在日常开发时,如果有文件上传下载的需求(比如用户头像),但是又不想使用对象存储,那么自己搭建一个 MinIO 服务器是一个比较简单的解决方案。

MinIO 是一个基于 Apache License v2.0 开源协议的对象存储服务。它兼容亚马逊S3云存储服务接口,非常适合于存储大容量非结构化的数据,例如图片、视频、日志文件、备份数据和容器/虚拟机镜像等,而一个对象文件可以是任意大小,从 几kb 到最大 5T 不等。

MinIO是一个非常轻量的服务,可以很简单的和其他应用的结合,类似 NodeJS, Redis 或者 MySQL。


MinIO 中文网站:
https://www.minio.org.cn/docs/minio/linux/operations/installation.html

部署 MinIO 并修改 minioadmin 的账户密码

本文主要介绍在 Ububtu 上单节点单硬盘部署 MinIO,步骤如下:

下载 MinIO 服务器

wget https://dl.min.io/server/minio/release/linux-amd64/minio

为 MinIO 二进制文件添加执行权限

chmod +x minio

在合适的位置创建一个文件夹,用于存储 MinIO 上传的文件数据

# 在根目录创建 minio-data 文件夹,存储 MinIO 上传的文件数据
mkdir ~/minio-data

安装 MinIO。将 MinIO 的二进制文件移到
/usr/local/bin/
目录下,以使其全局可用

sudo mv minio /usr/local/bin/

使用持久化环境变量作为 MinIO console 的登录账户和密码。

编辑
.bashrc
文件,这里使用 nano。

nano ~/.bashrc

在文件的最后加上环境变量

export MINIO_ROOT_USER=newrootuser
export MINIO_ROOT_PASSWORD=newrootpassword

重新加载
.bashrc
以使更改生效

source ~/.bashrc

启动 MinIO

nohup minio server --secure ~/minio-data

这里使用
nohup
确保会话关闭之后 MinIO 不会停止,也可以使用
screen
等会话技术,或者将 MinIO 作为一个服务启动。这里不做过多介绍。

启动 MinIO 之后,在浏览器访问 ip+9000 端口即可访问 MinIO 的 Web 控制台。如果访问不了,
请先检查 MinIO 的两个端口是否已经开放
,一个是 MinIO 的 WebUI 端口,这个是随机的;一个是 MinIO 的 API 端口,这个固定是 9000。

在浏览器访问 ip+9000 实际会跳转到 WebUI 的端口。但是在使用 API 上传下载文件需要使用 9000 端口。


给 MinIO 配置域名

如果不想直接暴露 MinIO 的地址和端口,则可以使用 Nginx 给 MinIO 配置域名访问。

在此之前,您需要先准备一个已备案的域名并解析到当前服务器。

步骤如下:

先安装 Nginx (如果没有的话)

sudo apt-get update
sudo apt-get install nginx

一般不建议直接更改位于
/etc/nginx/nginx.conf
的 nginx 主配置文件。采用如下的配置方式:


/etc/nginx/sites-available/
创建新的配置文件,可以直接以当前配置的域名未文件名。我这里由于需要配置多个域名,文件名叫做
minio.conf

cd /etc/nginx/sites-available/
touch minio.conf

书写配置文件

nano minio.conf

配置文件示例

# WebUI 配置
server {
    listen 80;
    server_name yourdomain.com; # 替换为您的域名

    location / {
        proxy_pass http://localhost:44366;  # 替换为实际的端口
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

# API 配置
server {
    listen 80;
    server_name api.yourdomain.com; # 替换为您的API域名

    location / {
        proxy_pass http://localhost:9000;   # 替换为实际的端口
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

启用新创建的配置文件。将
/etc/nginx/sites-available/minio.conf
链接到
/etc/nginx/sites-enabled/
目录

sudo ln -s /etc/nginx/sites-available/minio.conf /etc/nginx/sites-enabled/

可以先检查一下 nginx,确认没有语法错误

nginx -t

重启 nginx 服务

sudo systemctl restart nginx

此时,在本机浏览器上应该可以用域名访问 Minio console 了。


MinIO 调用示例

在 SpringBoot 中调用

先添加依赖

<dependencies>
    <dependency>
        <groupId>io.minio</groupId>
        <artifactId>minio</artifactId>
        <version>8.4.3</version>
    </dependency>
</dependencies>

然后创建一个服务类来实现文件的上传下载

import io.minio.*;
import io.minio.messages.Item;

import java.io.InputStream;
import java.security.SecureRandom;
import java.util.Base64;
import java.util.UUID;

@Service
public class MinIOService {

    private final MinioClient minioClient;

    public MinIOService() {
        try {
            // 配置更改成自己的,建议写在配置文件中
            this.minioClient = MinioClient.builder()
                    .endpoint("http://localhost:9000")
                    .credentials("minioadmin", "minioadmin")
                    .build();
        } catch (MinioException e) {
            throw new RuntimeException("Failed to create MinioClient", e);
        }
    }

    public String uploadFile(String bucketName, String objectName, InputStream stream, long size) {
        try {
            minioClient.putObject(bucketName, objectName, stream, size);
            return "File uploaded successfully.";
        } catch (MinioException | IOException e) {
            throw new RuntimeException("File upload failed.", e);
        }
    }

    public InputStream downloadFile(String bucketName, String objectName) {
        try {
            return minioClient.getObject(bucketName, objectName);
        } catch (MinioException | IOException e) {
            throw new RuntimeException("File download failed.", e);
        }
    }
}

并为其书写对应的 Controller

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.InputStreamResource;
import org.springframework.core.io.Resource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.io.InputStream;

@RestController
@RequestMapping("/minio")
public class MinIOController {

    @Autowired
    private MinIOService minIOService;

    @PostMapping("/upload")
    public String uploadFile(@RequestParam("file") MultipartFile file) {
        String bucketName = "my-bucket";
        String objectName = UUID.randomUUID().toString();
        try (InputStream stream = file.getInputStream()) {
            return minIOService.uploadFile(bucketName, objectName, stream, file.getSize());
        } catch (IOException e) {
            return "Failed to upload file.";
        }
    }

    @GetMapping("/download/{objectName}")
    public ResponseEntity<Resource> downloadFile(@PathVariable String objectName) {
        String bucketName = "my-bucket";
        InputStream stream = minIOService.downloadFile(bucketName, objectName);
        return ResponseEntity.ok()
                .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"" + objectName + "\"")
                .body(new InputStreamResource(stream));
    }
}

在 Flask 中调用

下载依赖

pip install minio

然后创建一个 Flask 来集成 MinIO

from flask import Flask, request, send_file, jsonify
from minio import Minio
from minio.error import S3Error

app = Flask(__name__)

# MinIO配置
minio_client = Minio(
    # 更改成自己的
    "localhost:9000",
    access_key="minioadmin",
    secret_key="minioadmin",
    secure=False
)

@app.route('/upload', methods=['POST'])
def upload_file():
    file = request.files['file']
    bucket_name = 'my-bucket'
    object_name = file.filename
    try:
        with open('/tmp/' + object_name, 'wb') as f:
            f.write(file.read())
        minio_client.fput_object(bucket_name, object_name, '/tmp/' + object_name)
        return jsonify({'message': 'File uploaded successfully'}), 200
    except S3Error as exc:
        return jsonify({'error': str(exc)}), 500

@app.route('/download/<object_name>', methods=['GET'])
def download_file(object_name):
    bucket_name = 'my-bucket'
    try:
        response = minio_client.get_object(bucket_name, object_name)
        return send_file(
            response.stream,
            as_attachment=True,
            attachment_filename=object_name,
            mimetype=response.headers['content-type']
        )
    except S3Error as exc:
        return jsonify({'error': str(exc)}), 500

if __name__ == '__main__':
    app.run(debug=True)

推荐阅读

MinIO Plus:https://mp.weixin.qq.com/s/kSkC3X-SQqo5GzXt-H66xw

首先编写一个android测试程序

功能:校验用户名和注册码,成功则弹出注册成功提示

以下仅给出关键部分的代码

res/layout/activity_main.xml

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
    android:layout_width="fill_parent"
    android:layout_height="fill_parent"
    android:orientation="vertical">

    <TextView
        android:id="@+id/textView1"
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:gravity="center"
        android:text="@string/info"
        android:textSize="20dp" />

    <LinearLayout
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:layout_marginLeft="10dp"
        android:orientation="horizontal">

        <TextView
            android:layout_width="wrap_content"
            android:layout_height="wrap_content"
            android:text="@string/username" />

        <EditText
            android:id="@+id/edit_username"
            android:layout_width="0dp"
            android:layout_height="wrap_content"
            android:layout_marginLeft="10dp"
            android:layout_marginRight="10dp"
            android:layout_weight="1"
            android:ems="10"
            android:hint="@string/hint_username"></EditText>
    </LinearLayout>

    <LinearLayout
        android:layout_width="match_parent"
        android:layout_height="wrap_content"
        android:layout_marginLeft="10dp"
        android:orientation="horizontal">

        <TextView
            android:layout_width="wrap_content"
            android:layout_height="wrap_content"
            android:text="@string/sn" />

        <EditText
            android:id="@+id/edit_sn"
            android:layout_width="0dp"
            android:layout_height="wrap_content"
            android:layout_marginLeft="10dp"
            android:layout_marginRight="10dp"
            android:layout_weight="1"
            android:ems="10"
            android:hint="@string/hint_sn"></EditText>
    </LinearLayout>

    <Button
        android:id="@+id/button_register"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_gravity="right"
        android:layout_marginRight="10dp"
        android:text="@string/register" />

</LinearLayout>

java/com/example/myapplication/MainActivity.java

package com.example.myapplication;

import android.app.Activity;
import android.os.Bundle;
import android.view.Menu;
import android.view.View;
import android.view.View.OnClickListener;
import android.widget.Button;
import android.widget.EditText;
import android.widget.Toast;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;

public class MainActivity extends Activity {
    private EditText edit_userName;
    private EditText edit_sn;
    private Button btn_register;

    @Override
    public void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        setTitle(R.string.unregister);  //模拟程序未注册
        edit_userName = (EditText) findViewById(R.id.edit_username);
        edit_sn = (EditText) findViewById(R.id.edit_sn);
        btn_register = (Button) findViewById(R.id.button_register);
        btn_register.setOnClickListener(new OnClickListener() {

            public void onClick(View v) {
                if (!checkSN(edit_userName.getText().toString().trim(),
                        edit_sn.getText().toString().trim())) {
                    Toast.makeText(MainActivity.this,       //弹出无效用户名或注册码提示
                            R.string.unsuccessed, Toast.LENGTH_SHORT).show();
                } else {
                    Toast.makeText(MainActivity.this,       //弹出注册成功提示
                            R.string.successed, Toast.LENGTH_SHORT).show();
                    btn_register.setEnabled(false);
                    setTitle(R.string.registered);  //模拟程序已注册
                }
            }
        });
    }

    @Override
    public boolean onCreateOptionsMenu(Menu menu) {
        getMenuInflater().inflate(R.menu.activity_main, menu);
        return true;
    }

    private boolean checkSN(String userName, String sn) {
        try {
            if ((userName == null) || (userName.length() == 0))
                return false;
            if ((sn == null) || (sn.length() != 16))
                return false;
            MessageDigest digest = MessageDigest.getInstance("MD5");
            digest.reset();
            digest.update(userName.getBytes());
            byte[] bytes = digest.digest();     //采用MD5对用户名进行Hash
            String hexstr = toHexString(bytes, ""); //将计算结果转化成字符串
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < hexstr.length(); i += 2) {
                sb.append(hexstr.charAt(i));
            }
            String userSN = sb.toString(); //计算出的SN
            //Log.d("crackme", hexstr);
            //Log.d("crackme", userSN);
            if (!userSN.equalsIgnoreCase(sn))   //比较注册码是否正确
                return false;
        } catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    private static String toHexString(byte[] bytes, String separator) {
        StringBuilder hexString = new StringBuilder();
        for (byte b : bytes) {
            String hex = Integer.toHexString(0xFF & b);
            if (hex.length() == 1) {
                hexString.append('0');
            }
            hexString.append(hex).append(separator);
        }
        return hexString.toString();
    }

}

2024-07-22T06:35:25.png

运行没有问题之后,通过
AndroidStudio
编译成apk文件

开始破解程序

破解 Android 程序通常的方法是将 apk 文件利用 ApkTool 反编译,生成 Smali 格式的反汇编代码,然后阅读 Smali 文件的代码来理解程序的运行机制,找到程序的突破口进行修改,最后使用 ApkTool 重新编译生成 apk 文件并签名,最后运行测试,如此循环,直至程序被成功破解。

使用
apk-tool
反编译apk程序

下载地址:
https://down.52pojie.cn/Tools/Android_Tools/apktool_2.9.3.jar

执行

java -jar apktool_2.9.3.jar d -f app-debug.apk -o output

2024-07-22T06:45:41.png

smail
目录下存放了程序的所有反汇编代码,
res
目录下则是程序的所有资源文件

先通过
res\values\string.xml
定位程序的错误信息

<resources>
...
    <string name="unsuccessed">无效用户名或注册码</string>
...
</resources>

再通过同目录下的
public.xml
找到
name="unsuccessed"
的id

<public type="string" name="unsuccessed" id="0x7f1000a5" />

通过该id可以到
smail
目录搜索,在
output\smali_classes3\com\example\myapplication\MainActivity$1.smali
搜索到一处结果

   91      iget-object v0, p0, Lcom/example/myapplication/MainActivity$1;->this$0:Lcom/example/myapplication/MainActivity;
   92  
   93:     const v2, 0x7f1000a5
   94  
   95      invoke-static {v0, v2, v1}, Landroid/widget/Toast;->makeText(Landroid/content/Context;II)Landroid/widget/Toast;

onclick方法


# virtual methods
.method public onClick(Landroid/view/View;)V
    .locals 3
    .param p1, "v"    # Landroid/view/View;

    .line 31
    iget-object v0, p0, Lcom/example/myapplication/MainActivity$1;->this$0:Lcom/example/myapplication/MainActivity;

    invoke-static {v0}, Lcom/example/myapplication/MainActivity;->access$000(Lcom/example/myapplication/MainActivity;)Landroid/widget/EditText;

    move-result-object v1

    invoke-virtual {v1}, Landroid/widget/EditText;->getText()Landroid/text/Editable;

    move-result-object v1

    invoke-virtual {v1}, Ljava/lang/Object;->toString()Ljava/lang/String;

    move-result-object v1

    invoke-virtual {v1}, Ljava/lang/String;->trim()Ljava/lang/String;

    move-result-object v1

    iget-object v2, p0, Lcom/example/myapplication/MainActivity$1;->this$0:Lcom/example/myapplication/MainActivity;

    .line 32
    invoke-static {v2}, Lcom/example/myapplication/MainActivity;->access$100(Lcom/example/myapplication/MainActivity;)Landroid/widget/EditText;

    move-result-object v2

    invoke-virtual {v2}, Landroid/widget/EditText;->getText()Landroid/text/Editable;

    move-result-object v2

    invoke-virtual {v2}, Ljava/lang/Object;->toString()Ljava/lang/String;

    move-result-object v2

    invoke-virtual {v2}, Ljava/lang/String;->trim()Ljava/lang/String;

    move-result-object v2

    .line 31
    invoke-static {v0, v1, v2}, Lcom/example/myapplication/MainActivity;->access$200(Lcom/example/myapplication/MainActivity;Ljava/lang/String;Ljava/lang/String;)Z

    move-result v0

    const/4 v1, 0x0

    if-nez v0, :cond_0 # 关键条件判断

    .line 33
    iget-object v0, p0, Lcom/example/myapplication/MainActivity$1;->this$0:Lcom/example/myapplication/MainActivity;

    const v2, 0x7f1000a5

    invoke-static {v0, v2, v1}, Landroid/widget/Toast;->makeText(Landroid/content/Context;II)Landroid/widget/Toast;

    move-result-object v0

    .line 34
    invoke-virtual {v0}, Landroid/widget/Toast;->show()V

    goto :goto_0

    .line 36
    :cond_0
    iget-object v0, p0, Lcom/example/myapplication/MainActivity$1;->this$0:Lcom/example/myapplication/MainActivity;

    const v2, 0x7f1000a2

    invoke-static {v0, v2, v1}, Landroid/widget/Toast;->makeText(Landroid/content/Context;II)Landroid/widget/Toast;

    move-result-object v0

    .line 37
    invoke-virtual {v0}, Landroid/widget/Toast;->show()V

    .line 38
    iget-object v0, p0, Lcom/example/myapplication/MainActivity$1;->this$0:Lcom/example/myapplication/MainActivity;

    invoke-static {v0}, Lcom/example/myapplication/MainActivity;->access$300(Lcom/example/myapplication/MainActivity;)Landroid/widget/Button;

    move-result-object v0

    invoke-virtual {v0, v1}, Landroid/widget/Button;->setEnabled(Z)V

    .line 39
    iget-object v0, p0, Lcom/example/myapplication/MainActivity$1;->this$0:Lcom/example/myapplication/MainActivity;

    const v1, 0x7f100099

    invoke-virtual {v0, v1}, Lcom/example/myapplication/MainActivity;->setTitle(I)V

    .line 41
    :goto_0
    return-void
.end method

经过分析,
if-nez v0, :cond_0
判断决定了校验是否通过,这句代码的意思是:如果v0不为0则跳转到cond_0,也就是注册失败的分支。

破解方法:将
if-nez
改为
if-eqz
也就是等于则为真

修改后保存,执行如下代码重新编译

java -jar apktool_2.9.3.jar b output

编译后需要对apk进行签名,这里我本地生成了一个测试签名

keytool -genkey -alias testalias -keyalg RSA -keysize 2048 -validity 36500 -keystore test.keystore

然后用360加固助手的工具包进行签名。

接下来可以通过adb命令安装和启动

adb install app-debug_sign.apk
adb shell am start -n com.example.myapplication/.MainActivity

2024-07-22T08:00:54.png

至此破解就算完成了。

使用IDA pro破解

因为使用apktool每次都需要重新编译,很花时间,idapro提供了快速测试的方法。

下载地址:
https://down.52pojie.cn/Tools/Disassemblers/IDA_Pro_v8.3_Portable.zip

用压缩包工具打开
app-debug.apk
,提取出
classes.dex
文件,通过ida pro打开

注意:如果
classes.dex
没有可能在
classes3.dex

按照先找错误信息的思路,按
alt+t
打开文本搜索功能,搜索:0x7f1000a5

2024-07-22T08:35:50.png

定位到关键代码(按空格可以切换视图)

2024-07-23T01:31:23.png

可以看到分支判断位于
CODE:000006BE
,将光标放在
if-nez
处,点击
hex-view
,修改
if-nez
的字节码

39 00 0F 00
改为
38 00 0F 00

然后关闭
ida pro
,不需要保存到database


c32asm
打开
classes3.dex
定位到
000006BE
,将39改为38,保存退出

2024-07-23T01:41:13.png

这里由于修改了dex文件,会导致dex文件在验证计算checksum会失败,从而导致程序安装失败,因此需要重新计算checksum值

用Dexfixer将classes3.dex文件checksum值修复

2024-07-23T01:51:10.png

将修复好的
classes3.dex
,重新拉入apk

aapt r app-debug.apk classes3.dex
aapt a app-debug.apk classes3.dex

删除META-INT(可使用winrar工具),并重新签名apk即可破解成功。

image

参考:《Android软件安全与逆向分析》

前言

监控k8s集群,目前主流就是使用prometheus以及其周围的生态,本文开始介绍怎么一步步完成k8s监控的建设

环境准备

组件 版本
操作系统 Ubuntu 22.04.4 LTS
minikube v1.30.1
docker 24.0.7
prometheus v2.54.1
kube-state-metrics v2.13.0
node-exporter v1.8.2

下载编排文件

本文所有的编排文件,
都在这里

▶ cd /tmp && git clone git@github.com:wilsonchai8/installations.git && cd installations/prometheus

使用minikube搭建k8s测试环境

1)下载
minikube

2)启动,
minikube start

▶ minikube start
▶ docker ps | grep minikube
db877d660750   kicbase/stable:v0.0.39     "/usr/local/bin/entr…"   37 seconds ago   Up 33 seconds  127.0.0.1:32782->22/tcp, 127.0.0.1:32781->2376/tcp, 127.0.0.1:32780->5000/tcp, 127.0.0.1:32779->8443/tcp, 127.0.0.1:32778->32443/tcp   minikube

3)检查k8s是否正常工作

▶ kubectl get node
NAME       STATUS   ROLES           AGE     VERSION
minikube   Ready    control-plane   4m41s   v1.26.3

这里需要注意一下,由于现在镜像地址全部被墙了,大家可以尝试这个方法解决

安装prometheus

1)创建命名空间

▶ kubectl create ns prometheus
namespace/prometheus created

2)启动

▶ cd /tmp/installations/prometheus
▶ kubectl apply -f prometheus.yaml

检查是否启动

▶ kubectl -n prometheus get pod -owide
NAME                                 READY   STATUS    RESTARTS   AGE     IP           NODE       NOMINATED NODE   READINESS GATES
prometheus-deploy-8495dfd557-xcwnp   1/1     Running   0          2m52s   10.244.0.3   minikube   <none>           <none>

这里需要注意的是configmap的配置,由于是一步一步的演示,所以configmap的文件内容是一步一步的充实进去的,后面会慢慢修改

apiVersion: v1
kind: ConfigMap
metadata:
  name: prometheus-cm
  labels:
    name: prometheus-cm
  namespace: prometheus
data:
  prometheus.yml: |-
    global:
      scrape_interval: 5s
      evaluation_interval: 5s

    alerting:
      alertmanagers:
        - static_configs:
            - targets: ['alertmanager:9093']

    rule_files:
      - /etc/prometheus/*.rules

    scrape_configs:
      - job_name: 'prometheus'
        static_configs:
        - targets: ['localhost:9090']

3)访问页面服务

查看service配置,nodeport的端口是32648

▶ kubectl -n prometheus get svc
NAME                 TYPE       CLUSTER-IP      EXTERNAL-IP   PORT(S)          AGE
prometheus-service   NodePort   10.99.231.160   <none>        9090:32648/TCP   12m
▶ kubectl get node -owide
NAME       STATUS   ROLES           AGE    VERSION   INTERNAL-IP    EXTERNAL-IP   OS-IMAGE             KERNEL-VERSION     CONTAINER-RUNTIME
minikube   Ready    control-plane   4d5h   v1.26.3   192.168.49.2   <none>        Ubuntu 20.04.5 LTS   6.8.0-45-generic   docker://23.0.2

访问
192.168.49.2:32648

至此,prometheus已经安装完成了,监控的框架算是搭建起来了,但是没有监控指标,这并不是一个完整的监控系统,我们需要prometheus监控k8s的基础指标

安装k8s exporter

简单来说,exporter就是提供监控数据的组件,prometheus定期到exporter采集数据。而即将介绍的
kube-state-metrics
,就是专门用来提供k8s相关数据的exporter

而安装exporter也是非常简单的,exporter也是一个组件服务,只需要把它编排进k8s即可

▶ cd /tmp/installations/prometheus
▶ kubectl apply -f kube-state-metrics.yaml

查看exporter是否启动

▶ kubectl -n kube-system get pod | grep kube-state-metrics
kube-state-metrics-6cd66dbcd8-4mqh4   1/1     Running   0             63s

exporter安装完成,需要告诉prometheus去采集新的exporter,修改prometheus configmap

apiVersion: v1
kind: ConfigMap
metadata:
  name: prometheus-cm
  labels:
    name: prometheus-cm
  namespace: prometheus
data:
  prometheus.yml: |-
    global:
      scrape_interval: 5s
      evaluation_interval: 5s

    alerting:
      alertmanagers:
        - static_configs:
            - targets: ['alertmanager:9093']

    rule_files:
      - /etc/prometheus/*.rules

    scrape_configs:
      - job_name: 'prometheus'
        static_configs:
        - targets: ['localhost:9090']

# 从这里是新加的
      - job_name: "prometheus-kube-state-metrics"
        static_configs:
          - targets: ["kube-state-metrics.kube-system:8080"]

修改了configmap之后,需要重启prometheus

▶ kubectl -n prometheus rollout restart deploy prometheus-deploy

重新打开页面之后查看,有新的监控指标被采集上来了

安装node exporter

需要注意的是,
kube-state-metrics
并没有关注node相关的监控数据,这时候又有一个exporter需要上场了,那就是
node-exporter

▶ cd /tmp/installations/prometheus
▶ kubectl apply -f node-exporter.yaml

查看exporter是否启动成功

▶ kubectl -n prometheus get pod | grep node-exporter
node-exporter-q8rmq                  1/1     Running   0          75s

node-exporter
的工作方式与
kube-state-metrics
不一样,
kube-state-metrics
是借助k8s的服务发现能力,可以知道k8s集群内部到底有多少pod、deploy、service等资源的状态

node-exporter
也是部署在k8s内部,通过daemonset的方式,在每一个节点启动一个采集服务,然后暴露api等待prometheus来采集数据,但是
node-exporter
自身并没有服务发现能力,所以在节点扩容或者缩容的时候,prometheus并不知道当前有多少节点需要采集。虽然prometheus不知道,但是k8s确知道当前集群有多少节点,所以这里
node-exporter
也需要借用k8s得服务发现的能力来自动发现当前的节点数

修改prometheus configmap

apiVersion: v1
kind: ConfigMap
metadata:
  name: prometheus-cm
  labels:
    name: prometheus-cm
  namespace: prometheus
data:
  prometheus.yml: |-
    global:
      scrape_interval: 5s
      evaluation_interval: 5s

    alerting:
      alertmanagers:
        - static_configs:
            - targets: ['alertmanager:9093']

    rule_files:
      - /etc/prometheus/*.rules

    scrape_configs:
      - job_name: 'prometheus'
        static_configs:
        - targets: ['localhost:9090']

      - job_name: "prometheus-kube-state-metrics"
        static_configs:
          - targets: ["kube-state-metrics.kube-system:8080"]

# 从这里是新加的
      - job_name: 'kubernetes-nodes'
        tls_config:
          ca_file: /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
        bearer_token_file: /var/run/secrets/kubernetes.io/serviceaccount/token
        kubernetes_sd_configs:
        - role: node
        relabel_configs:
        - source_labels: [__address__]
          regex: '(.*):10250'
          replacement: '${1}:9100'
          target_label: __address__
          action: replace
        - action: labelmap
          regex: __meta_kubernetes_node_label_(.+)

修改了configmap之后,需要重启prometheus

▶ kubectl -n prometheus rollout restart deploy prometheus-deploy

这里已经可以看到node相关的指标了

切到
Status --> Targets
,可以看到通过k8s服务发现的节点

为了验证服务发现,我们新加一个节点

▶ minikube node add

检查新加节点

▶ kubectl get node
NAME           STATUS   ROLES           AGE    VERSION
minikube       Ready    control-plane   5d3h   v1.26.3
minikube-m02   Ready    <none>          70s    v1.26.3

再去页面检查,已经自动发现了第二个节点

至此,一个可以监控k8s各种资源的监控系统已经初步达成

注意事项

由于本文是演示一步一步安装prometheus以及相关组件,所以配置文件也是一步一步累加出来的,最终呈现在仓库的配置文件是最终版,可以直接将编排文件fully apply,从而跳过这些调试步骤

联系我

  • 联系我,做深入的交流


至此,本文结束
在下才疏学浅,有撒汤漏水的,请各位不吝赐教...

ReentrantLock的条件队列是实现“等待通知”机制的关键,之前在《
java线程间通信:等待通知机制
》一文中讲过了使用ReentrantLock实现多生产者、多消费者的案例,这个案例实际上在java源码的注释中已经给了,可以看Condition接口上的注释中相关的代码:

class BoundedBuffer {
    final Lock lock = new ReentrantLock();
    final Condition notFull  = lock.newCondition();
    final Condition notEmpty = lock.newCondition();

    final Object[] items = new Object[100];
    int putptr, takeptr, count;

    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length)
                notFull.await();
            items[putptr] = x;
            if (++putptr == items.length) putptr = 0;
            ++count;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0)
                notEmpty.await();
            Object x = items[takeptr];
            if (++takeptr == items.length) takeptr = 0;
            --count;
            notFull.signal();
            return x;
        } finally {
            lock.unlock();
        }
    }
}

在文章《
详解AQS二:ReentrantLock公平锁原理
》中已经详细说了lock和unlock方法的实现原理,实际上就是利用了AQS队列实现阻塞、加锁、解锁。那么当lock、unlock方法中间夹杂着Condition的await、signal方法的调用,又发生了什么事情呢?

一、条件等待的原理:ConditionObject

一切都要从ConditionObject类说起,它是Condition接口的实现类,lock.newCondition方法调用实际返回的就是ConditionObject类的实例。那么
condition.await
方法调用和
condition.signal
方法调用到底发生了什么呢?这实际上就是AQS队列中的Node元素和条件等待队列中元素的相互移动:

image-20241222203731545

复习一下lock和unlock的句式

private final Lock lock=new ReentrantLock(); // 创建一个Lock接口实例
……
// 申请锁lock,如果发生竞争且竞争锁失败,则当前线程进入AQS队列等待
lock.lock(); 
try{
  // 阻塞被释放后,当前线程执行临界区代码
    //await、signal方法调用
  ……
}finally{
  // 在finally块中释放锁,当前线程节点从AQS队列中移除
  lock.unlock(); 
}

await方法、signal方法都必须在临界区代码中执行,await方法调用之后大家都猜到了:当前线程会进入条件等待队列等待。但是明明当前线程还在AQS队列中,还没有调用unlock方法呢,那条件等待队列中和AQS队列中是不是都会有当前线程的等待实例?答案是否定的。实际上调用sinal方法之后当前线程会创建新Node等待在Condition队列尾部,同时释放AQS锁,本身节点将会在AQS队列中被移除。

同样的,当调用signal方法之后,Condition队列中的节点会被移除,同时会创建新节点到AQS队列中等待重新抢占锁。

二、await方法原理

await方法执行时当前线程在AQS队列中必然是头部节点,也就是已经获取锁的节点。await方法执行的原理就是将节点从AQS队列中的头部挪到Condition队列的尾部,之后释放锁并唤醒AQS队列的下一个节点让其抢占锁。

public final void await() throws InterruptedException {
    //如果当前线程发生了中断,就抛出中断异常
    if (Thread.interrupted())
        throw new InterruptedException();
    //将当前线程封装Node节点并加到Condition等待队列
    Node node = addConditionWaiter();
    //释放锁并获取释放锁时查询的锁状态值(重入次数),以便于以后重新进入AQS队列使用
    int savedState = fullyRelease(node);
    //中断状态暂存标记
    int interruptMode = 0;
    /*
     * 循环查询节点是否在AQS队列中,如果在AQS队列中表示已经重新进入AQS队列了,
     * 这意味着有别的线程调用了signal方法唤醒了当前线程
     */
    while (!isOnSyncQueue(node)) {
        //不在AQS队列中,那就继续挂起等待
        LockSupport.park(this);
        //检查中断状态,防止由于发生中断导致LockSupport.park(this);失效
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    //抢锁,如果在抢锁过程中发生了异常,则将中断标记设置为REINTERRUPT
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    //如果是因中断导致的LockSupport.park(this);挂起失效,则遍历等待队列中的节点,删除无效节点
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    //判定是否应当抛出中断异常还是仅仅恢复中断标记
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

await方法的代码延续了AQS的代码风格,非常精简,但是信息量很大,每一行代码都值得推敲。总体来说,await方法干了以下的事情:

(1)封装当前线程为新Node节点,并添加到等待队列尾部

(2)AQS队列释放锁,并唤醒AQS同步队列中头部节点的后继节点

(3)执行while循环,将该节点的线程阻塞,直到该节点离开等待队列,进入AQS同步队列;或者检测到了中断异常结束while循环。

(4)退出while循环后,执行acquireQueued方法尝试获取锁。

(5)执行善后工作,遍历等待队列中的节点删除无效节点。

(6)最后根据中断发生的时机,signal方法调用前,就抛出异常;否则就设置下中断标记,是否抛出异常看业务代码处理。

1、加入等待队列:addConditionWaiter

/**
 * 添加一个新的等待节点到等待队列
 * @return 添加到等待队列中的节点对象
 */
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // 如果最后一个节点取消等待了,则执行unlinkCancelledWaiters方法遍历等待队列删除无效节点
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    //封装当前线程为新Node节点,并且状态为CONDITION
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;//将新节点放到队尾
    lastWaiter = node;
    return node;
}

这个方法很简单,就是创建了新节点Node对象,并且置为CONDTION状态表示等待在等待队列中;最后将其放到等待队列的队尾。

等待队列是一个单向队列,节点和节点之间使用nextWaiter指针链接。

值得一提的是,等待队列中节点Node类和AQS同步队列中的Node类是同一个类,之前在分析AQS时没使用到的节点状态Node.CONDITION以及属性nextWaiter在等待队列中全都用到了。

addConditionWaiter方法中比较让人费解的是这段代码:

Node t = lastWaiter;
// 如果最后一个节点取消等待了,则执行unlinkCancelledWaiters方法遍历等待队列删除无效节点
if (t != null && t.waitStatus != Node.CONDITION) {
    unlinkCancelledWaiters();
    t = lastWaiter;
}

我们知道等待在等待队列中的节点必定是CONDITION状态,若不是这个状态,表示该节点已经取消等待了,那这时候就要将该节点从等待队列中移除。我们会有疑问,为什么该节点取消等待的时候不自己出队?感觉AQS所有的操作都似乎有“Lazy”的特性,包括AQS队列的初始化是在自旋入队的时候、AQS节点的出队在自旋抢占锁的时候。。都不是自己主动操作,而是“事件驱动”模式的。

2、删除无效节点:unlinkCancelledWaiters

节点入等待队列的时候可能会执行unlinkCancelledWaiters方法,该方法的作用是删除等待队列中已经取消等待的节点。

private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) {
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            if (trail == null){
                firstWaiter = next;
            }else{
                trail.nextWaiter = next;
            }
            if (next == null){
                lastWaiter = trail;
            }
        }else{
            trail = t;
        }
        t = next;
    }
}

这段代码通过trail和t两个指针一前一后遍历整个等待队列,并剔除掉非Node.CONDITION状态的节点。

unlinkCancelledWaiters方法整体来说比较简单,该方法仅仅在当前线程持有锁的时候被调用,用于将取消等待的节点从等待队列中移除。在以下两个场景下该方法会被调用:

  • 在条件等待期间线程取消等待
  • 最后一个等待节点已经取消等待的前提下,插入新的等待节点(addConditionWaiter方法内的unlinkCancelledWaiters方法调用就是因为此)

该方法需要避免在signal方法没有被调用的情况下产生无法被回收的垃圾节点,全遍历等待队列中的所有节点也是无奈之举,好在该方法被调用有前提:sinal方法没有被调用,而且等待节点因为一些原因取消了等待。

该方法遍历了等待队列中的所有节点,这样做有个好处:如果有大量的等待节点取消了等待,可能会导致“取消风暴”,这样一次遍历就取消所有指向垃圾节点的指针,可以避免多次重新遍历以提高运行效率。

3、释放锁:fullRelease

节点加入等待队列以后会释放锁,并唤醒后继节点

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        //获取锁状态,实际上就是当前线程的锁重入次数,以方便后续恢复
        int savedState = getState();
        //释放锁
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

ReentrantLock公平锁和非公平锁锁释放都调用的同一个方法:release方法,fullRelease方法一直没有被提及,因为它是为等待队列准备的方法。fullRelease方法内部会调用release方法。

fullRelease方法和release方法最大的区别就是release方法只会返回true/false表示释放锁成功了还是失败了,fullRelease方法则会先保存锁的状态,释放完锁以后会将该state返回给调用方await方法,用于以后恢复线程在AQS同步队列中的状态。

4、是否在AQS队列:isOnSyncQueue

释放完成锁以后会执行while循环,不断执行isOnSyncQueue方法以判断节点是否在同步队列中,如果节点出现在了AQS同步队列中,说明节点已经被唤醒了,它可以重新尝试获取锁了。

final boolean isOnSyncQueue(Node node) {
    /**
     * 节点状态为CONDITION表示节点肯定在等待队列
     * AQS同步队列是双向队列,等待在AQS同步队列中的所有节点prev都有值,
     * 为null表示不在AQS同步队列中等待;反过来node.prev不为空并不表示node节点
     * 一定在AQS同步队列。
     */
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    /**
     * 等待队列中的节点通过nextWaiter指针相互链接,next指针用于AQS队列
     * next指针不为空表示节点必定在AQS同步队列;反过来,node.next为空说明不了
     * node节点不在AQS队列
     */
    if (node.next != null) 
        return true;
	
    //从尾部到头部遍历节点
    return findNodeFromTail(node);
}

/**
 * 从尾部到头部遍历AQS同步队列查找指定节点
 */
private boolean findNodeFromTail(Node node) {
    Node t = tail;
    for (;;) {
        if (t == node)
            return true;
        if (t == null)
            return false;
        t = t.prev;
    }
}

得解释一下:

node.prev不为空说明不了node节点在AQS同步队列:

可以看下AQS的enq自旋入队方法

image-20241226155745270

可以看到,在CAS操作入队之前,node.prev就已经设置了值了,而CAS可能会失败,所以就算node.prev有值,node节点也可能没有在队列中。

node.next为空说明不了node节点不在AQS同步队列:

这个很好理解,AQS同步队列tail节点的next指针是null。

最后一个问题是:为什么要从尾部到头部遍历AQS同步队列查找指定节点,而不是从头部到尾部查找?

一个重要的原因是当前节点刚加入AQS队列尾部的话,从尾部查找可能不会经历很多遍历就能查找到该节点,从尾部开始查找可以提高效率。

三、await方法中的中断

上一章节讲到了isOnSyncQueue方法,该方法判定当前节点是否在AQS同步队列中,如果在同步队列中,则结束while循环,开始执行acquireQueue方法抢占锁;否则执行LockSupport.park方法将线程阻塞。看似逻辑很简单,但是涉及到了中断相关的逻辑,比较复杂,需要单独拎出来掰扯掰扯。

public final void await() throws InterruptedException {
    ....省略之前的代码....
    while (!isOnSyncQueue(node)) {
            //不在AQS队列中,那就继续挂起等待
            LockSupport.park(this);//①
            //检查中断状态,防止由于发生中断导致LockSupport.park(this);失效
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)//②
                break;
    }
    //抢锁,如果在抢锁过程中发生了异常,则将中断标记设置为REINTERRUPT
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    //如果是因中断导致的LockSupport.park(this);挂起失效,则遍历等待队列中的节点,删除无效节点
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    //判定是否应当抛出中断异常还是仅仅恢复中断标记
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

1、LockSupport.park放行的原因

在①处执行了LockSupport.park代码,线程被阻塞挂起,那什么时候线程会被释放继续运行临界区代码呢?

有三个可能的原因:

  1. signal方法被调用后,线程节点自旋进入AQS同步队列,正常等待机会获取锁之后被前驱节点线程调用LockSupport.unpark方法唤醒

  2. 有别的线程调用了signal方法,signal方法中可能会调用LockSupport.unpark方法直接唤醒线程

  3. 线程发生了中断,LockSupport.lock方法阻塞失效了。

可能很多人都会想到signal方法会唤醒线程,但是想不到可能发生的中断会让LockSupport.lock方法失效仍然会释放线程让代码继续运行。这也是为什么接下来的代码要紧接着检查线程中断。

2、检查中断:checkInterruptWhileWaiting

确切的说是检查是否发生了中断,以及如果发生了中断,是在signal方法执行前发生了中断,还是signal方法执行后发生了中断。

/**
 * 返回值有三种:
 * 0:没有发生中断
 * -1(THROW_IE):发生了中断,而且中断是在signal方法调用前发生的
 * 1(REINTERRUPT):发生了中断,而且中断是在signal方法调用后发生的
 */
private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}

/**
 * 如果需要,在取消等待后将节点转移到AQS同步队列。
 * 如果节点是在被唤醒前取消等待的,则返回true,否则返回false。
 * 只有发生了中断才会调用该方法。
 */
final boolean transferAfterCancelledWait(Node node) {
    
    /*
     * 这里的CAS操作如果成功,说明当前节点的状态是Node.CONDITION
     * 也就是说,当前节点必定是在等待队列,更进一步,说明了当前中断
     * 是在被唤醒之前发生的,因为如果调用了signal方法,当前节点必然
     * 已经在AQS同步队列中,已经是状态0了。
     */
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        //自旋入队
        enq(node);
        return true;
    }
    /*
     * 代码执行到这里,并不能说明signal方法已经执行完毕,有可能signal方法
     * 正在执行中,Node状态已经更改成了0,但是enq方法还没执行完成,节点还在
     * 疯狂自旋入队。所以这里要判断是否完成了enq方法,就要判定下当前节点是否
     * 已经在AQS同步队列中。如果enq方法还没执行完,就让出CPU时间片,稍稍等待
     * 下,通常enq方法会很快完成,所以不用担心这里会浪费CPU资源。
     */
    while (!isOnSyncQueue(node))
        Thread.yield();
    //singal方法执行后发生的中断,返回false
    return false;
}

checkInterruptWhileWaiting方法有三个返回值:0、1(REINTERRUPT)、-1(THROW_IE)。在这里,要特别注意
Thread.interrupted()
方法的调用,该方法在上一章节《
详解AQS二:ReentrantLock公平锁原理
》就说过了,它有个很重要的特点是会重置中断状态为false,并且返回中断状态。checkInterruptWhileWaiting的返回值1和-1都代表发生过中断。不管是1还是-1,都会停止while循环,中断值则会暂存在
interruptMode
变量中。

while (!isOnSyncQueue(node)) {
        //不在AQS队列中,那就继续挂起等待
        LockSupport.park(this);//①
        //检查中断状态,防止由于发生中断导致LockSupport.park(this);失效
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)//②
            break;
}

3、抢锁以及重设中断状态

中断循环以后,当前线程就开始抢锁,抢锁方法是acquireQueued方法,该方法已经在文章《
详解AQS二:ReentrantLock公平锁原理
》中讲过,不再赘述。

acquireQueued方法返回值是true/false,true表示发生了中断,false表示未发生中断,这个中断是在抢锁过程中发生的;在抢锁之前,while循环中调用了checkInterruptWhileWaiting方法,该方法调用了
Thread.interrupted()
方法重置了中断状态,正是为了不影响acquireQueued方法的调用。

我们分析下抢锁代码

//抢锁,如果在抢锁过程中发生了异常,则将中断标记设置为REINTERRUPT
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
    interruptMode = REINTERRUPT;

这段代码执行了如下代码流程:

  1. 执行acquireQueued方法抢锁,抢锁成功后返回中断标记
  2. 如果执行acquireQueued方法抢锁的过程中发生了中断,则判断之前在等待队列中是否发生了中断以及中断类型是否是THROW_IE
  3. 如果当前节点在等待队列中等待的时候发生了中断并且中断类型是THROW_IE,则将中断标记interruptMode修改为REINTERRUPT。

现在我们聚焦于
interruptMode != THROW_IE
这个判断条件,如果interruptMode不是THROW_IE,它会是什么值?只有可能是剩下的两种值:

  • 0:在等待队列中未发生中断,但是acquireQueued方法执行的时候发生了中断

  • 1(REINTERRUPT):在等待队列中等待的时候发生了中断,而且中断是在signal方法调用后发生的,而且之后在acquireQueued方法中抢锁的时候也发生了中断

在以上两种情况下,interruptMode会被修改成REINTERRUPT值(实际上第二种情况interruptMode的值已经是REINTERRUPT,重复修改也无妨)

这样,最终interruptMode一共只有可能有三种类型的值:

  • 0:在等待队列中未发生中断,在执行acquireQueued方法抢锁的过程中也没发生中断
  • 1(REINTERRUPT):有可能在等待队列中未发生中断,但是在执行acquireQueued方法抢锁的过程中发生了中断;也有可能在等待队列中发生了中断,而且中断是在signal方法调用后发生的,同时在执行acquireQueued方法抢锁的过程中也发生了中断
  • -1(THROW_IE):在等待队列中发生了中断,而且中断是在signal方法调用前发生的;是否在执行acquireQueued方法抢锁的过程中发生了中断无法得知。

针对这三种类型的值,接下来会如何处理呢?

4、中断的处理方式

await方法最后一段代码:

//判定是否应当抛出中断异常还是仅仅恢复中断标记
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);

这里首先判断了interruptMode是否是0,只有非0的值才会被处理,也就是说,只有异常类型才会被处理。

private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();
}

终于看到了对THROW_IE和REINTERRUPT类型的处理:

THROW_IE
:抛出InterruptedException异常

REINTERRUPT
:设置中断标记,这样如果在接下来的业务处理中出现了sleep等等待方法,将会抛出InterruptedException异常。

使用表格总结一下异常处理的各种情况

image-20241226214525606

可以看到,是否要抛出异常还是设置中断标记,主要取决于节点在等待队列中发生中断的时机:

等待队列中在signal方法调用

发生的中断:

一定会抛出中断异常
,不管acquireQueued方法有没有发生中断

等待队列中在signal方法调用

发生的中断:

一定会设置中断标记位
,不管acquireQueued方法有没有发生中断

等待队列中没有发生中断:
如果抢锁过程中发生了中断,就设置中断标记位;如果抢锁过程中没有发生中断,就什么都不做。

思考一下,为什么要这么做呢?

acquireQueued是一个“uninterruptible”的类,也就是说,它不会抛出中断异常,但是它会将中断反馈给调用方,让调用方决定该如何处理异常,那await就作为调用方,结合了在等待队列中是否发生异常做出了综合决定:一切看是否在等待队列中发生的异常种类,如果没有发生异常,就向AQS同步队列处理异常类型一样,让业务自己决定是否抛出异常;如果在等待队列中发生了异常:

  • 在signal方法调用前发生异常,说明临界区代码没打算执行,直接抛出中断异常中断临界区代码执行即可
  • 在signal方法调用后发生异常,说明临界区代码本打算执行,临界区代码自己都没强制停止执行,是否要抛出中断异常停止临界区代码执行让业务代码自己判断去吧。

5、删除无效节点:unlinkCancelledWaiters

await方法最后会判定node.nextWaiter是否为null,如果是null则遍历等待队列中的节点,删除无效节点

//如果是因中断导致的LockSupport.park(this);挂起失效,则遍历等待队列中的节点,删除无效节点
if (node.nextWaiter != null) // clean up if cancelled
    unlinkCancelledWaiters();

为什么会判断node.nextWaiter是否为null?想要知道这个答案,得看看signal方法的实现逻辑。

四、signal方法原理

signal方法就是把线程节点从等待队列转移到AQS同步队列,之后就是AQS正常获取锁并被前置节点线程唤醒后继续执行同步代码块中await方法之后的的临界区代码的过程。

public final void signal() {
    //如果当前锁的持有线程不是当前线程,就抛出异常
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    //firstWaiter是等待时间最久的线程,从它开始唤醒最公平
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

1、判定锁的持有线程:isHeldExclusively

这个方法是个AQS的钩子方法,实现在ReentrantLock类中的Sync抽象类

protected final boolean isHeldExclusively() {
    return getExclusiveOwnerThread() == Thread.currentThread();
}

这个方法特别简单,就是简单判定了下当前线程是否是持有锁的线程,什么情况下会发生不是持有锁的线程呢?明明当前线程正在运行中,它肯定是持有锁的线程啊,举个简单的例子如下所示:

/**
 * @author kdyzm
 * @date 2024/12/27
 */
public class AQSNotHolderThread {
    
    public static void main(String[] args) {
        Lock lock1 = new ReentrantLock();
        Lock lock2 = new ReentrantLock();
        Condition con1 = lock1.newCondition();
        
        Thread thread = new Thread(() -> {
            lock1.lock();
            try {
                con1.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock1.unlock();
            }
        });
        thread.start();

        lock2.lock();
        try {
            //在lock2的临界区代码使用lock1的con1,会抛出IllegalMonitorStateException异常
            con1.signal();
        } finally {
            lock2.unlock();
        }
    }
}

以上代码会抛出
IllegalMonitorStateException
异常。

isHeldExclusively方法调用的目的就是要保证Condition方法的调用必须在对应的锁下的同步代码块中。

2、doSignal方法

doSignal方法会从头部开始查找第一个没有被取消的节点,将其转移到AQS同步队列或者直接唤醒线程。

private void doSignal(Node first) {
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        //不管后续有没有transferForSignal成功,都将nextWaiter指针置为null
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

在上面的代码中,
first.nextWaiter = null;
这块代码是亮点,虽然不经意的一行代码,却是解释第三章第五节:删除无效节点:unlinkCancelledWaiters 的关键,之前留下了一个疑问,在await方法的最后:

//如果是因中断导致的LockSupport.park(this);挂起失效,则遍历等待队列中的节点,删除无效节点
if (node.nextWaiter != null) // clean up if cancelled
    unlinkCancelledWaiters();

会判定node.nextWaiter是否为null,如果不是null,就执行unlinkCancelledWaiters方法清除无效节点。

从doSignal方法中可以看到,只要调用了调用了doSignal方法,节点的nextWaiter就会被置为null;那await方法中在LockSupport.park方法调用之后发现了node.nextWaiter竟然不为空,那就表示signal方法没有被调用,换句话说,是因为中断才使得LockSupport.park失效最终导致代码运行到这里的,在signal方法调用前发生了中断,那节点是取消等待状态,需要将节点从等待队列中移除出去,执行
unlinkCancelledWaiters()
方法就合理了。

3、转移节点到AQS:transferForSignal

该方法的主要目的是将线程节点从条件等待队列转移到AQS同步队列。

final boolean transferForSignal(Node node) {
    /*
     * 如果节点已经取消等待,则不再尝试转移节点到AQS同步队列
     */
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    /*
     * 节点自旋转进入AQS同步队列,并返回前置节点
     */
    Node p = enq(node);
    int ws = p.waitStatus;
    /**
     * ws>0:前置节点状态是取消状态
     * !compareAndSetWaitStatus(p, ws, Node.SIGNAL):设置前驱节点为Signal状态失败
     */
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);//唤醒节点线程
    return true;
}

我们分析一下这段代码:

本来目标节点在await方法中已经是被LockSupport.park方法将线程阻塞了,enq方法完成之后,实际上该节点已经完成了入队,应该是是等待前驱节点唤醒它的状态,似乎等价于AQS的acquire方法

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

中的
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
方法调用,然而真的等价吗?实际上不一定等价,其区别就是前驱节点的状态,由于目标节点是“
先阻塞,后入队
”,和正常进入AQS队列等待的线程“
先入队,后阻塞
”的流程是反着来的,所以它没有调用过
shouldParkAfterFailedAcquire
方法校正前驱节点状态:

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

shouldParkAfterFailedAcquire方法执行完成后会将取消状态的节点剔除,并且将前置节点状态修正为SIGNAL。

在transferForSignal方法中的代码

if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);//唤醒节点线程

实际上就是在快速验证前置节点状态,如果前置节点状态错误且无法修正,则将目标线程直接唤醒让其继续执行await方法中LockSupport.park方法之后的代码逻辑:

while (!isOnSyncQueue(node)) {
        //不在AQS队列中,那就继续挂起等待
        LockSupport.park(this);//①
        //检查中断状态,防止由于发生中断导致LockSupport.park(this);失效
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)//②
            break;
}
//抢锁,如果在抢锁过程中发生了异常,则将中断标记设置为REINTERRUPT
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
    interruptMode = REINTERRUPT;

由于在signal方法中已经入队,所以这里会结束while循环,执行下一个if语句中的
acquireQueued(node, savedState)
,这个方法中如果获取锁失败,它会调用
shouldParkAfterFailedAcquire
方法:

/**
 * 已经在队列中的线程通过独占模式获取锁的方法,该方法不受中断的影响。
 * 该方法也同样用于等待在条件等待队列中的线程获取锁。
 */
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            //如果前置节点是头部节点,当前节点就尝试抢占锁
            if (p == head && tryAcquire(arg)) {
                //抢锁成功后将抢锁节点设置为头结点
                setHead(node);
                //释放头结点利于GC
                p.next = null; 
                failed = false;
                return interrupted;
            }
            //如果应该阻塞等待就挂起线程进入阻塞状态
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

调用
shouldParkAfterFailedAcquire
方法就会重新校正前置节点状态为SIGNAL并且剔除掉前置节点中的取消状态的节点。

这样await方法将会在acquireQueued方法调用处阻塞,等待获取锁之后执行后续的临界区代码。

这样就完美闭环了。



最后,欢迎关注我的博客呀:

https://blog.kdyzm.cn

END.