τα στοιχεία επικοινωνίας μου
Ταχυδρομείο[email protected]
2024-07-12
한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina
Όπως είναι κατανοητό επί του παρόντος, το KafkaListener του Spring Kafka χρησιμοποιεί SimpleAsyncTaskExecutor, αλλά δεν χρησιμοποιείται σε άλλα σενάρια.
Χωρίς ταυτόχρονο περιορισμό ρεύματος, κάθε φορά που εκτελείται το SimpleAsyncTaskExecutor.execute(runnable), θα δημιουργείται ένα νέο νήμα για την ασύγχρονη εκτέλεση της εργασίας.
/**
* 不带并发限流控制的 SimpleAsyncTaskExecutor
*/
public static void main(String[] args) {
SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("my-test-");
Runnable runnable = new Runnable() {
@SneakyThrows
@Override
public void run() {
Thread.sleep(1000L);
System.out.println("当前线程: " + Thread.currentThread().getName());
}
};
for (int i = 0; i < 10; i++) {
executor.execute(runnable);
}
}
Εκτυπώστε ως εξής:
当前线程: my-test-4
当前线程: my-test-10
当前线程: my-test-5
当前线程: my-test-3
当前线程: my-test-9
当前线程: my-test-7
当前线程: my-test-2
当前线程: my-test-6
当前线程: my-test-8
当前线程: my-test-1
Εκτελέστε ταυτόχρονο περιορισμό ρεύματος Κάθε φορά που εκτελείται το SimpleAsyncTaskExecute(runnable), θα δημιουργείται ένα νέο νήμα για την ασύγχρονη εκτέλεση εργασιών.
Ποια είναι λοιπόν η διαφορά μεταξύ περιορισμού ταυτόχρονου ρεύματος και περιορισμού μη ταυτόχρονου ρεύματος;
/**
* 带并发限流控制的 SimpleAsyncTaskExecutor
*/
public static void main(String[] args) {
SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor("my-test-");
// 并发线程限制
executor.setConcurrencyLimit(3);
Runnable runnable = new Runnable() {
@SneakyThrows
@Override
public void run() {
Thread.sleep(1000L);
System.out.println("当前线程: " + Thread.currentThread().getName());
}
};
for (int i = 0; i < 10; i++) {
executor.execute(runnable);
}
// 会发现主线程被卡住，因为在 SimpleAsyncTaskExecutor 中会阻塞等待
System.out.println("主线程执行完成");
}
Εκτυπώστε ως εξής:
当前线程: my-test-3
当前线程: my-test-1
当前线程: my-test-2
------------------------------------------
当前线程: my-test-6
当前线程: my-test-5
当前线程: my-test-4
------------------------------------------
当前线程: my-test-8
当前线程: my-test-7
当前线程: my-test-9
------------------------------------------
主线程执行完成
当前线程: my-test-10
Ο πηγαίος κώδικας του SimpleAsyncTaskExecutor είναι σχετικά μικρός, ας δούμε απευθείας αυτήν την κλάση.
public class SimpleAsyncTaskExecutor extends CustomizableThreadCreator
implements AsyncListenableTaskExecutor, Serializable {
/**
* -1 表示不进行并发限流
*/
public static final int UNBOUNDED_CONCURRENCY = -1;
/**
* 0 表示其他线程等待其他线程执行完
*/
public static final int NO_CONCURRENCY = 0;
// 并发限流的实现对象
// 并发限流就是靠这个类实现的
private final ConcurrencyThrottleAdapter concurrencyThrottle = new ConcurrencyThrottleAdapter();
@Nullable
private ThreadFactory threadFactory;
@Nullable
private TaskDecorator taskDecorator;
public SimpleAsyncTaskExecutor() {
super();
}
public SimpleAsyncTaskExecutor(String threadNamePrefix) {
super(threadNamePrefix);
}
/**
* 设置并发线程数
* 给 concurrencyThrottle 的 concurrencyLimit 字段设值
* 默认 concurrencyThrottle 的 concurrencyLimit 的值为 -1，表示不进行并发限流
*/
public void setConcurrencyLimit(int concurrencyLimit) {
this.concurrencyThrottle.setConcurrencyLimit(concurrencyLimit);
}
/**
* 当前是否是并发限流的
* 其实就是看 concurrencyThrottle 的 concurrencyLimit >= 0 ?
*/
public final boolean isThrottleActive() {
return this.concurrencyThrottle.isThrottleActive();
}
/**
* 常用的 execute()
*/
@SuppressWarnings("deprecation")
@Override
public void execute(Runnable task) {
execute(task, TIMEOUT_INDEFINITE);
}
/**
* 执行给定的 task
*/
@Deprecated
@Override
public void execute(Runnable task, long startTimeout) {
Assert.notNull(task, "Runnable must not be null");
Runnable taskToUse = (this.taskDecorator != null ?
this.taskDecorator.decorate(task) : task);
// 1. 如果需要进行并发限流，走下面的逻辑
if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
this.concurrencyThrottle.beforeAccess();
doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
}
else {
// 2. 不需要并发限流，直接执行 doExecute(task)
doExecute(taskToUse);
}
}
protected void doExecute(Runnable task) {
// 1. 直接创建一个新的线程！！！
// 这就是为什么 SimpleAsyncTaskExecutor 每次执行 execute() 都会创建一个新线程的原因
Thread thread = (this.threadFactory != null ?
this.threadFactory.newThread(task) : createThread(task));
// 2. 调用 thread.start() 异步执行任务
thread.start();
}
/**
* ConcurrencyThrottleAdapter 只有两个方法，都是由父类实现的
*/
private static class ConcurrencyThrottleAdapter extends ConcurrencyThrottleSupport {
@Override
protected void beforeAccess() {
super.beforeAccess();
}
@Override
protected void afterAccess() {
super.afterAccess();
}
}
/**
* 包装了 Runnable 对象，并且本身也是 Runnable 对象
* 装饰器模式
*/
private class ConcurrencyThrottlingRunnable implements Runnable {
private final Runnable target;
public ConcurrencyThrottlingRunnable(Runnable target) {
this.target = target;
}
@Override
public void run() {
try {
this.target.run();
}
finally {
// 主要是在 finally 块中执行 concurrencyThrottle.afterAccess()
concurrencyThrottle.afterAccess();
}
}
}
}
Μέσα από την παραπάνω περιγραφή, έχουμε μια απλή κατανόηση του SimpleAsyncTaskExecutor.
Χωρίς έλεγχο ταυτόχρονης ροής, είναι εύκολο να γίνει κατανοητό ότι ένα νέο νήμα θα δημιουργείται κάθε φορά που εκτελείται η SimpleAsyncTaskExecutor.execute().
Εξετάζουμε κυρίως πώς εκτελεί τον έλεγχο ροής όταν εκτελείται ταυτόχρονος έλεγχος ροής.
// ---------------------- SimpleAsyncTaskExecutor ------------------------
public void execute(Runnable task, long startTimeout) {
Runnable taskToUse = (this.taskDecorator != null ?
this.taskDecorator.decorate(task) : task);
// 1. 如果需要进行并发限流，走下面的逻辑
if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
// 1.1 执行 this.concurrencyThrottle.beforeAccess()
// 如果被并发限流的话会阻塞等待
this.concurrencyThrottle.beforeAccess();
// 1.2 此时没有被限流住
// 将 task 包装为 ConcurrencyThrottlingRunnable
// ConcurrencyThrottlingRunnable 的 run() 的 finally 块会释放资源
// 使其他线程能通过限流
doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
}
else {
// 2. 不需要并发限流，直接执行 doExecute(task)
doExecute(taskToUse);
}
}
Το this.concurrencyThrottle.beforeAccess() και το ConcurrencyThrottlingRunnable.run() είναι και τα δύο βασικά σημεία.
Μπορείτε να δείτε ότι ελέγχει το όριο συγχρονισμού μέσω συγχρονισμένου και concurrencyLimit.
// ---------------------- ConcurrencyThrottleSupport ------------------------
protected void beforeAccess() {
if (this.concurrencyLimit == 0) {
// 不允许 concurrencyLimit == 0
throw new IllegalStateException();
}
// 1. 存在并发限流的场景，this.concurrencyLimit > 0
if (this.concurrencyLimit > 0) {
// 2. 尝试获取 monitor 对象锁，获取不到的话在这里阻塞，等其他线程释放锁
synchronized (this.monitor) {
// 3. 如果当前并发线程 >= this.concurrencyLimit
// 当前线程 wait 等待，直到其他线程唤醒它
while (this.concurrencyCount >= this.concurrencyLimit) {
this.monitor.wait();
}
// 4. 当前并发线程数 concurrencyCount++
this.concurrencyCount++;
}
}
}
Ας ρίξουμε μια ματιά στο run() αυτής της κλάσης.
// --------------------- ConcurrencyThrottlingRunnable -----------------------
private class ConcurrencyThrottlingRunnable implements Runnable {
private final Runnable target;
public ConcurrencyThrottlingRunnable(Runnable target) {
this.target = target;
}
@Override
public void run() {
try {
// 1. 执行目标 target.run()
this.target.run();
}
finally {
// 2. 执行 concurrencyThrottle.afterAccess()
concurrencyThrottle.afterAccess();
}
}
}
// ---------------------- ConcurrencyThrottleSupport ------------------------
protected void afterAccess() {
// 并发限流场景下
// 先获取 monitor 对象锁，执行 concurrencyCount--，再唤醒 wait 中的线程
if (this.concurrencyLimit >= 0) {
synchronized (this.monitor) {
this.concurrencyCount--;
this.monitor.notify();
}
}
}
Σε αυτό το σημείο, ολοκληρώνεται η ανάλυση του SimpleAsyncTaskExecutor.
Στην πραγματική ανάπτυξη, δεν πρέπει να χρησιμοποιούμε το SimpleAsyncTaskExecutor για να αποφύγουμε καταστροφικά προβλήματα.
Έχει αφοσιωθεί στην έρευνα της τεχνολογίας για περισσότερα από 30 χρόνια και είναι ικανός σε διάφορες γλώσσες όπως java, linux, javascript, php, css κ.λπ. Έχει κάνει πολλές συνεισφορές στον τομέα του ανοιχτού κώδικα σταθμός τεκμηρίωσης προγραμματιστή για να μοιραστείτε ορισμένα ζητήματα στην ανάπτυξη τεχνολογίας για μελλοντική αναφορά
Ταχυδρομείο[email protected]