Multi-threading in Java

What is Multi-threading?

Multi-threading refers to the capability of a program or process to execute multiple sequences of instructions concurrently. In computer science, a thread is the smallest sequence of programmed instructions that can be managed independently by an operating system's scheduler.

Threads are a fundamental component of multitasking and allow programs to perform multiple tasks or processes seemingly simultaneously. Instead of having a single path of execution, a program with multiple threads can perform various activities concurrently. Each thread runs independently and shares the resources of the parent process, such as memory, open files, and other system resources.

Why do we need Multi-threading?

​ From the picture above, we can understand why we need multi-threading. The reason is that multi-threading can improve performance. If one processor only can run one thread, the other threads will have no choices, they need to wait. We can imagine a ticket center as a processor so if the ticket center only has one staff, people can buy tickets in only one queue, but if the ticket center has five staff, we can buy a ticket from any of them, and it will be more faster.

​ Nowadays, Multi-threading is crucial in modern computing. It has some advantages below:

1. Improved Performance: Utilizing multiple threads allows a program to execute various tasks simultaneously, which can significantly enhance performance. For example, in applications that involve heavy computations or I/O-bound operations, dividing work among multiple threads can reduce processing time.
2. Concurrency: Multi-threading enables concurrent execution of tasks, facilitating the handling of multiple operations at the same time. This capability is particularly beneficial in applications requiring responsiveness, such as user interfaces that need to remain interactive while performing background tasks.
3. Utilizing Multi-Core Processors: Most modern computers feature multi-core processors. Multi-threading allows a program to take advantage of these cores by running different threads on separate cores, achieving parallelism and better resource utilization.
4. Improved Responsiveness: Separating time-consuming tasks from the main thread ensures that the application remains responsive. For instance, in software applications, moving lengthy operations to separate threads prevents the main thread (UI thread) from being blocked.
5. Asynchronous Operations: Multi-threading facilitates asynchronous processing. For tasks such as network communication, file I/O, or waiting for external events, asynchronous execution using threads allows other parts of the program to continue running without being held up by these operations.
6. Resource Sharing: Threads within a process share the same memory space, which allows for efficient communication and data sharing among threads.
7. Complex System Implementations: In complex systems, multi-threading is necessary for managing multiple concurrent activities. For instance, web servers handling numerous requests simultaneously use threads to manage and process those requests concurrently.

How to use Multi-threading?

1. Create a new class that extends the Thread class and override its run() method. After that, you can create a new thread to run your logic code, but you have to manage the thread by yourself. For example, you can't create a new thread in a 1000-times run loop, it will make trouble. So you have to control the amount of thread.

   class MyThread extends Thread {
       public void run() {
           System.out.println("This is running in a separate thread.");
       }
   }
   
   public class Main {
       public static void main(String[] args) {
           MyThread thread = new MyThread();
           thread.start(); // Start the thread
       }
   }
   
   

2. Using a thread pool is more convenient for the developer, we can just create a pool with a limited thread amount, and then add a new thread to the thread pool. The thread pool will manage the thread amount. We can check the "runWorker" and "execute" method in ThreadPoolExecutor.java.

   final void runWorker(Worker w) {
           Thread wt = Thread.currentThread();
           Runnable task = w.firstTask;
           w.firstTask = null;
           w.unlock(); // allow interrupts
           boolean completedAbruptly = true;
           try {
               while (task != null || (task = getTask()) != null) {
                   w.lock();
                   // If pool is stopping, ensure thread is interrupted;
                   // if not, ensure thread is not interrupted.  This
                   // requires a recheck in second case to deal with
                   // shutdownNow race while clearing interrupt
                   if ((runStateAtLeast(ctl.get(), STOP) ||
                        (Thread.interrupted() &&
                         runStateAtLeast(ctl.get(), STOP))) &&
                       !wt.isInterrupted())
                       wt.interrupt();
                   try {
                       beforeExecute(wt, task);
                       Throwable thrown = null;
                       try {
                           task.run();
                       } catch (RuntimeException x) {
                           thrown = x; throw x;
                       } catch (Error x) {
                           thrown = x; throw x;
                       } catch (Throwable x) {
                           thrown = x; throw new Error(x);
                       } finally {
                           afterExecute(task, thrown);
                       }
                   } finally {
                       task = null;
                       w.completedTasks++;
                       w.unlock();
                   }
               }
               completedAbruptly = false;
           } finally {
               processWorkerExit(w, completedAbruptly);
           }
       }
   
   public void execute(Runnable command) {
           if (command == null)
               throw new NullPointerException();
           /*
            * Proceed in 3 steps:
            *
            * 1. If fewer than corePoolSize threads are running, try to
            * start a new thread with the given command as its first
            * task.  The call to addWorker atomically checks runState and
            * workerCount, and so prevents false alarms that would add
            * threads when it shouldn't, by returning false.
            *
            * 2. If a task can be successfully queued, then we still need
            * to double-check whether we should have added a thread
            * (because existing ones died since last checking) or that
            * the pool shut down since entry into this method. So we
            * recheck state and if necessary roll back the enqueuing if
            * stopped, or start a new thread if there are none.
            *
            * 3. If we cannot queue task, then we try to add a new
            * thread.  If it fails, we know we are shut down or saturated
            * and so reject the task.
            */
           int c = ctl.get();
           if (workerCountOf(c) < corePoolSize) {
               if (addWorker(command, true))
                   return;
               c = ctl.get();
           }
           if (isRunning(c) && workQueue.offer(command)) {
               int recheck = ctl.get();
               if (! isRunning(recheck) && remove(command))
                   reject(command);
               else if (workerCountOf(recheck) == 0)
                   addWorker(null, false);
           }
           else if (!addWorker(command, false))
               reject(command);
       }
   

   ​ Then after checking the code, we can understand the thread pool is helping us to manage the threads. It will try to create and execute a new thread in the pool, but if the pool is full, it will try to put the task in the queue, if the queue is full, it will reject the task, and the run loop will keep checking the task and execute it when the current thread is free.

​ Here is a demo using ThreadpoolExecutor.

package com.javademo.thread;

import java.util.concurrent.*;

public class ThreadPoolDemo {

    //创建核心线程数5，最大线程数10的线程池，观察线程输出了解线程池的工作原理：
//    提交一个任务到线程池中，线程池的处理流程如下：
//    1、判断线程池里的核心线程是否都在执行任务，如果不是（核心线程空闲或者还有核心线程没有被创建）则创建一个新的工作线程来执行任务。如果核心线程都在执行任务，则进入下个流程。
//    2、线程池判断工作队列是否已满，如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。
//    3、判断线程池里的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。
    //线程池优点：
//    1.能有效利用空闲线程；
//    2.避免过度创建新线程导致超过CPU线程数而崩溃；
    public static void main(String[] args) {

        //生成工作队列对象，声明容量，当线程超过核心线程数时会放入工作队列等待，当工作队列满了才开辟新线程
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(5);

        //生成饱和策略，饱和策略主要是声明当线程数超过最大线程数时应该怎么做
        RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy(); //是强制停止程序抛出异常，正常不会使用这个策略
        RejectedExecutionHandler rejectedExecutionHandler1 = new ThreadPoolExecutor.CallerRunsPolicy(); //继续等待使用线程池里的线程执行
        RejectedExecutionHandler rejectedExecutionHandler2 = new ThreadPoolExecutor.DiscardOldestPolicy(); //丢弃最旧的工作任务，以执行当前任务
        RejectedExecutionHandler rejectedExecutionHandler3 = new ThreadPoolExecutor.DiscardPolicy(); //不处理，丢弃该任务

        //生成线程池
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, queue, rejectedExecutionHandler1);

        //执行线程
        int totalThreadNum = 20;
        for (int i = 0; i < totalThreadNum; i++) {
            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
          
            threadPool.execute(thread);
            System.out.println("线程池中活跃线程数：" + threadPool.getPoolSize());
            System.out.println("工作任务队列等待执行任务数：" + queue.size());
        }

        //关闭线程池
        threadPool.shutdown();
    }
}

How to make the thread-safe?

​ In a multi-threaded environment, several threads might attempt to access and modify shared resources concurrently. Without proper synchronization or coordination mechanisms, concurrent access to shared resources can lead to issues such as race conditions, data corruption, and inconsistent states.

​ Here are some ways to make sure the thread is safe.

1. Using Synchronized Lock

   //使用syncronized，可以将线程里的synchronized注释以观察区别
           String lockName = "lock";
           Thread thread1 = new Thread(() -> {
               synchronized (lockName){
                   for (int i = 0; i < 5; i++) {
                       System.out.println("李四");
                       try {
                           Thread.sleep(100);
                       } catch (InterruptedException e) {
                           e.printStackTrace();
                       }
                   }
               }
   
           });
           thread1.start();
   
           Thread thread2 = new Thread(() -> {
               synchronized (lockName){
                   for (int i = 0; i < 5; i++) {
                       System.out.println("张三");
                       try {
                           Thread.sleep(100);
                       } catch (InterruptedException e) {
                           e.printStackTrace();
                       }
                   }
               }
           });
           thread2.start();
   

2. Using ReentrantLock

   //此处使用ReentrantLock的trylock()是尝试获取锁不会一直等待，如果使用lock()会一直等待
           final Lock lock = new ReentrantLock();
           Thread thread3 = new Thread(() -> {
               try {
                   boolean pass = lock.tryLock(1, TimeUnit.SECONDS);
                   if (pass){
                       try {
   //                        lock.lock();
                           for (int i = 0; i < 5; i++) {
                               System.out.println("李小龙");
                               Thread.sleep(100);
                           }
                       }finally {
                           lock.unlock();
                       }
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           });
           thread3.start();
   
           Thread thread4 = new Thread(() -> {
               try {
                   boolean pass = lock.tryLock(5, TimeUnit.SECONDS);
                   if (pass){
                       try {
   //                        lock.lock();
                           for (int i = 0; i < 5; i++) {
                               System.out.println("成龙");
                               Thread.sleep(100);
                           }
                       }finally {
                           lock.unlock();
                       }
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           });
           thread4.start();
   

3. Using Semaphore

   //使用semaphore，信号量设为1表示并发数为1，起到锁的作用
           Semaphore semaphore = new Semaphore(1);
           Thread thread5 = new Thread(() -> {
               try {
                   semaphore.acquire();
                   for (int i = 0; i < 5; i++) {
                       System.out.println("李连杰");
                       Thread.sleep(100);
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               } finally {
                   semaphore.release();
               }
           });
           thread5.start();
   
           Thread thread6 = new Thread(() -> {
               try {
                   semaphore.acquire();
                   for (int i = 0; i < 5; i++) {
                       System.out.println("甄子丹");
                       Thread.sleep(100);
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               } finally {
                   semaphore.release();
               }
           });
           thread6.start();
   

4. Using CAS (Compare and Swap)

   //非阻塞同步CAS,通常来说一个CAS接收三个参数，数据的现值V，进行比较的值A，准备写入的值B。只有当V和A相等的时候，才会写入B。无论是否写入成功，都会返回V
           CasCounter casCounter = new CasCounter(1);
           Thread thread7 = new Thread(() -> {
               try {
                   boolean casResult = casCounter.compareAndSet(1, 2);
                   if (casResult){
                       for (int i = 0; i < 5; i++) {
                           System.out.println("周星驰");
                           Thread.sleep(100);
                       }
                       casCounter.set(3);
                   }
   
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           });
           thread7.start();
   
           Thread thread8 = new Thread(() -> {
               try {
                   Thread.sleep(500); //睡眠线程等cascounter设置value为3
                   boolean casResult = casCounter.compareAndSet(3, 4);
                   if (casResult){
                       for (int i = 0; i < 5; i++) {
                           System.out.println("吴孟达");
                           Thread.sleep(100);
                       }
                   }
   
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           });
           thread8.start();
   

5. Using ThreadLocal

   //非同步方案，使用ThreadLocal共享变量，各个线程对其设置的变量仅其线程可见
           ThreadLocal<Integer> threadLocal = ThreadLocal.withInitial(() -> 1);
           Thread thread9 = new Thread(() -> {
               Integer sign = threadLocal.get();
               System.out.println("线程一的threadlocal值-1："+sign);
               threadLocal.set(2);
               sign = threadLocal.get();
               System.out.println("线程一的threadlocal值-2："+sign);
           });
           thread9.start();
   
           Thread thread10 = new Thread(() -> {
               try {
                   Thread.sleep(100);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
   
               Integer sign = threadLocal.get();
               System.out.println("线程二的threadlocal值-1："+sign);
               threadLocal.set(3);
               sign = threadLocal.get();
               System.out.println("线程二的threadlocal值-2："+sign);
               threadLocal.remove();
           });
           thread10.start();