基于多线程的方式优化 FLink 程序

本文涉及的产品
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云原生数据库 PolarDB MySQL 版,Serverless 5000PCU 100GB
云数据库 Redis 版,社区版 2GB
推荐场景:
搭建游戏排行榜
简介: 这篇内容介绍了线程的基本概念和重要性。线程是程序执行的最小单位,比进程更细粒度,常用于提高程序响应性和性能。多线程可以实现并发处理,利用多核处理器,实现资源共享和复杂逻辑。文章还讨论了线程的五种状态(NEW、RUNNABLE、BLOCKED、WAITING、TIMED_WAITING和TERMINATED)以及如何在Java中创建和停止线程。最后提到了两种停止线程的方法:使用标识和中断机制。

一、前言

线程算是相对较高级的内容,主要的原因不是说他难,而是它不可见。最近基于多线程的方式优化了一些 FLink 程序,所以这一系列,我们聊聊多线程

二 线程

2.1 进程和线程关系

进程是计算机系统进行资源分配和调度的最小单位,换句话说我们平时双击那些后缀为 .exe的文件时都会产生一个进程。

进程可以产生若干个线程,是程序执行的最小单位,换句话说,进程就是房子,线程就是房子内一个个干活的人

2.2 为什么需要线程

线程在计算机编程中扮演着重要角色,其重要性主要体现在以下几个方面:

  1. 提高程序响应性:通过多线程处理,程序可以变得更加灵活和响应更加及时。在一个单线程程序中,如果有一个耗时的操作,会导致整个程序阻塞,影响用户体验;而多线程可以使程序保持活跃,允许其他线程继续执行,从而提高程序的响应性。
  2. 提高程序性能:多线程可以充分利用多核处理器的优势,实现并发执行多个任务,加快程序运行速度,提高系统整体性能。通过并行执行,程序可以更有效地利用计算资源,加快任务完成的速度。
  3. 实现并发处理:多线程允许程序同时执行多个任务,这对于需要同时处理多个事件或任务的应用程序至关重要。例如,在服务器端应用中,多线程可以同时处理多个客户端请求。
  4. 资源共享:多个线程可以共享进程的资源,如内存空间、文件句柄等,这种资源共享有助于简化程序设计,并提高效率。不同线程之间可以相互通信、共享数据,协同工作来完成复杂任务。
  5. 实现复杂逻辑:有些程序需要同时进行多项任务,通过多线程可以更好地组织和管理复杂的逻辑,提高程序的可维护性和可拓展性。
  6. 实现异步编程:多线程可以实现异步操作和事件驱动,允许程序在等待某些操作完成时继续执行其他操作,提高程序的效率和灵活性。异步编程模型通过非阻塞方式进行任务处理,可以有效提升程序的吞吐量和性能。 综合以上原因,线程在计算机编程中是不可或缺的,它提供了一种有效的机制来实现并发处理、提高程序的响应性和性能、实现资源共享以及管理复杂的程序逻辑。因此,了解和灵活运用线程是提升程序效率和优化系统性能的重要手段。

2.3 线程的状态

java

复制代码

public enum State {
    /**
     * Thread state for a thread which has not yet started.
     */
    NEW,

    /**
     * Thread state for a runnable thread.  A thread in the runnable
     * state is executing in the Java virtual machine but it may
     * be waiting for other resources from the operating system
     * such as processor.
     */
    RUNNABLE,

    /**
     * Thread state for a thread blocked waiting for a monitor lock.
     * A thread in the blocked state is waiting for a monitor lock
     * to enter a synchronized block/method or
     * reenter a synchronized block/method after calling
     * {@link Object#wait() Object.wait}.
     */
    BLOCKED,

    /**
     * Thread state for a waiting thread.
     * A thread is in the waiting state due to calling one of the
     * following methods:
     * <ul>
     *   <li>{@link Object#wait() Object.wait} with no timeout</li>
     *   <li>{@link #join() Thread.join} with no timeout</li>
     *   <li>{@link LockSupport#park() LockSupport.park}</li>
     * </ul>
     *
     * <p>A thread in the waiting state is waiting for another thread to
     * perform a particular action.
     *
     * For example, a thread that has called <tt>Object.wait()</tt>
     * on an object is waiting for another thread to call
     * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
     * that object. A thread that has called <tt>Thread.join()</tt>
     * is waiting for a specified thread to terminate.
     */
    WAITING,

    /**
     * Thread state for a waiting thread with a specified waiting time.
     * A thread is in the timed waiting state due to calling one of
     * the following methods with a specified positive waiting time:
     * <ul>
     *   <li>{@link #sleep Thread.sleep}</li>
     *   <li>{@link Object#wait(long) Object.wait} with timeout</li>
     *   <li>{@link #join(long) Thread.join} with timeout</li>
     *   <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
     *   <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
     * </ul>
     */
    TIMED_WAITING,

    /**
     * Thread state for a terminated thread.
     * The thread has completed execution.
     */
    TERMINATED;
}

2.4 如何创建线程

java

复制代码

Thread thread = new Thread();
thread.start();

在Java中,当调用线程对象的start()方法时,实际上是调用了start0()方法,该方法会启动一个新的本地操作系统线程, 然后调用Java中的run()方法来执行线程的任务。所有 线程的主要工作的方法就是 run 方法,那么怎么样来丰富 run 方法的内容呢?

首先通过匿名内部类来启动线程,如:

java

复制代码

Thread thread1 = new Thread(){
    @Override
    public void run() {
        while (true){
            if (Thread.currentThread().isInterrupted()){
                System.out.println("Interruted!");
                break;
            }
            Thread.yield();
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println("thread1==");
        }
    }
} ;

thread1.start();

当然也可以实现 Runnable 接口来实现线程的功能

java

复制代码

public class CreateThread3 implements Runnable {
    public static void main(String[] args) {
        Thread t1 = new Thread(new CreateThread3());
        t1.start();
    }

    @Override
    public void run() {
        System.out.println("Oh,I am Runnable");
    }

}

2.5 线程停止

2.5.1 使用标识

java

复制代码

public class ControlledThread extends Thread {
    private volatile boolean shouldStop = false;

    public void stopThread() {
        shouldStop = true;
    }

    @Override
    public void run() {
        while (!shouldStop) {
            // 线程执行的任务
            System.out.println("Thread is running...");
        }
        System.out.println("Thread stopped.");
    }

    public static void main(String[] args) {
        ControlledThread thread = new ControlledThread();
        thread.start();

        // 模拟停止线程
        try {
            Thread.sleep(3000); // 等待3秒
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        thread.stopThread();
    }
}

需要注意的是,这里使用 volatile 变量来保证该变量对于任意线程可见,如果不用 volatile 的话,则线程可能会无法停止

2.5.2 使用 interrup()

java

复制代码

public class InterruptedThread extends Thread {

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            // 线程执行的任务
            System.out.println("Thread is running...");
        }
        System.out.println("Thread stopped.");
    }

    public static void main(String[] args) {
        InterruptedThread thread = new InterruptedThread();
        thread.start();

        // 模拟停止线程
        try {
            Thread.sleep(3000); // 等待3秒
            thread.interrupt(); // 中断线程
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}


转载来源:https://juejin.cn/post/7376099820509642786

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
19天前
|
缓存 Java
深入理解Java并发编程:线程池的应用与优化
【5月更文挑战第30天】本文将深入探讨Java并发编程中的一个重要主题——线程池。我们将详细解析线程池的概念、应用及其优化方法,帮助读者更好地理解和使用线程池,提高程序的性能和效率。
|
14天前
|
算法 安全 Java
Java性能优化(四)-多线程调优-Synchronized优化
JVM在JDK1.6中引入了分级锁机制来优化Synchronized,当一个线程获取锁时,首先对象锁将成为一个偏向锁,这样做是为了优化同一线程重复获取导致的用户态与内核态的切换问题;其次如果有多个线程竞争锁资源,锁将会升级为轻量级锁,它适用于在短时间内持有锁,且分锁有交替切换的场景;轻量级锁还使用了自旋锁来避免线程用户态与内核态的频繁切换,大大地提高了系统性能;但如果锁竞争太激烈了,那么同步锁将会升级为重量级锁。减少锁竞争,是优化Synchronized同步锁的关键。
30 2
|
1天前
|
流计算
实时计算 Flink版操作报错合集之如何优化资源分配
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
2天前
|
缓存 NoSQL 数据处理
实时计算 Flink版产品使用问题之读取数据太慢该如何优化
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2天前
|
SQL 数据处理 API
实时计算 Flink版产品使用问题之holo的io以及cpu使用较为稳定,sink端busy一直在20%左右,有时候50%,该如何优化
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4天前
|
缓存 算法 Java
深入解析线程上下文切换的原理与优化策略
深入解析线程上下文切换的原理与优化策略
10 0
|
7天前
|
并行计算 算法 C#
C# Mandelbrot和Julia分形图像生成程序更新到2010-9-14版 支持多线程计算 多核处理器
此文档是一个关于分形图像生成器的介绍,作者分享了个人开发的M-J算法集成及色彩创新,包括源代码和历史版本。作者欢迎有兴趣的读者留言交流,并提供了邮箱(delacroix_xu@sina.com)以分享资源。文中还展示了程序的发展历程,如增加了真彩色效果、圈选放大、历史记录等功能,并分享了几幅精美的分形图像。此外,还提到了程序的新特性,如导入ini文件批量输出图像和更新一批图片的功能。文档末尾附有多张程序生成的高分辨率分形图像示例。
|
13天前
|
监控 算法 Java
Java性能优化(九)-多线程调优-垃圾回收机制优化
Java性能优化(九)-多线程调优-垃圾回收机制优化
17 0
|
13天前
|
缓存 Java 测试技术
Java性能优化(八)-多线程调优-线程池大小设置
Java性能优化(八)-多线程调优-线程池大小设置
13 0
|
13天前
|
安全 Java 大数据
Java性能优化(七)-多线程调优-并发容器的使用
Java性能优化(七)-多线程调优-并发容器的使用
19 0