<> Implementation of multithreading
Implementing thread is the basis of concurrent programming , Because we have to implement threads first , Then you can continue a series of subsequent operations .
<> Basic implementation mode
<>Runable
public class ImplementRunable implements Runnable { @Override public void run()
{ while (true) { // The name of the output thread , Distinguish from main thread name System.out.println(Thread.currentThread().
getName()); try { // Thread sleep for one second Thread.sleep(1000); } catch (Exception e) { throw
new RuntimeException(e); } } } public static void main(String[] args) { new
Thread(new ImplementRunable()).start(); } }
Essentially still new Thread, It's just that Runable Object is passed in as a constructor
<>Thread
The most common is through inheritance Thread Class or directly new Thread object , But we know Java
Single inheritance , That is, if we inherit Thread Class cannot inherit other classes , So this is what we often use Runable Reasons for multithreading , At this time, we can Runable Object passed to Thread
Constructor for . Here is Thread Common constructors
Let's look at an example
public class ExtendsThread extends Thread { public ExtendsThread() { //
Sets the name of the current thread this.setName("MyThread"); } @Override public void run() { //
every other 1s Output the name of the current thread once while (true) { // The name of the output thread , Distinguish from main thread name System.out.println(Thread.
currentThread().getName()); try { // Thread sleep for one second Thread.sleep(1000); } catch (
Exception e) { throw new RuntimeException(e); } } } public static void main(
String[] args) { new ExtendsThread().start(); } }
<>Callable
Runnable and Callable All represent tasks to be executed in different threads .Runnable from JDK1.0 From the beginning ,Callable Yes JDK1.5 Increased .
Their main difference is Callable of
call() Method can return a value and throw an exception , and Runnable of run() Methods do not have these functions , and Callable You can return the data loaded with calculation results Future object .
public class ImplementCallable implements Callable<Integer> { @Override public
Integer call() throws Exception { return new Random().nextInt(); } public static
void main(String[] args) throws ExecutionException, InterruptedException {
// Create thread pool ExecutorService service = Executors.newFixedThreadPool(1); // Submit task , Combined use
Future Submit return results Future<Integer> future = service.submit(new ImplementCallable());
Integer integer = future.get(); System.out.println(integer); } }
But it should be noted that Callable It can't be directly connected with Thread
Used , It needs to be used with thread pool , We also demonstrated how to pass Future Object get results . In fact, here we can also Callable coordination FutureTask use , The demonstration is behind us FutureTask I'll show you when I can
<>TimerTask
public class TimerTaskDemo { /** * delay 100ms after , interval 1s Print out ：hello world * * @param
args * @throws InterruptedException */ public static void main(String[] args)
throws InterruptedException { Timer t = new Timer(); t.scheduleAtFixedRate(new
TimerTask() { @Override public void run() { System.out.println("hello world"); }
}, 100, 1000); } }
This looks like a new implementation , But when you go to see TimerTask When the implementation of , You will find that this class actually inherits from Runnable In other words, this is actually through Runnable Implementation mode , In fact, it is to TimerTask Object into a TaskQueue Object in the queue , Then wait until the scheduling time comes , Then execute TimerTask of run
method .
Need to pay attention to when we use Timer Class time , Prompt us to use ScheduledExecutorService To replace it , Why , that is because Timer Thread pool not used , And the whole Timer Scheduling is single threaded , So we still have to take a look Timer Specific implementation principle of
Let's sort out a task scheduling process
*
We created a Timer object
public Timer() { this("Timer-" + serialNumber()); } public Timer(String name) {
thread.setName(name); thread.start(); }
We see that a thread is started in this constructor , This thread is TimerThread
*
Used scheduleAtFixedRate method , A scheduled task was created and scheduled , In fact, I just added this task to TaskQueue In the queue
public void scheduleAtFixedRate(TimerTask task, Date firstTime, long period) {
if (period <= 0) throw new IllegalArgumentException("Non-positive period.");
sched(task, firstTime.getTime(), period); } private void sched(TimerTask task,
long time, long period) { if (time < 0) throw new IllegalArgumentException(
"Illegal execution time."); // Constrain value of period sufficiently to
prevent numeric // overflow while still being effectively infinitely large. if (
Math.abs(period) > (Long.MAX_VALUE >> 1)) period >>= 1; synchronized(queue) { if
(!thread.newTasksMayBeScheduled) throw new IllegalStateException("Timer already
cancelled."); synchronized(task.lock) { if (task.state != TimerTask.VIRGIN)
throw new IllegalStateException( "Task already scheduled or cancelled"); task.
nextExecutionTime= time; task.period = period; task.state = TimerTask.SCHEDULED;
} // take task Added to the queue of the timer , queue.add(task); if (queue.getMin() == task) queue.
notify(); } }
*
Timer Realized it NB
Scheduling function of , We are not involved in this part , Belongs to black box , In fact, it's not black , Let's uncover it , The secret is TimerThread inside , We know that the thread object is Timer It started when it was created
private TaskQueue queue; TimerThread(TaskQueue queue) { this.queue = queue; }
// Thread entry , run method public void run() { try { // Call core method —— loop mainLoop(); } finally {
// Someone killed this Thread, behave as if Timer cancelled synchronized(queue)
{ newTasksMayBeScheduled = false; queue.clear(); // Eliminate obsolete
references } } } /** * The main timer loop. (See class comment.) */ private void
mainLoop() { while (true) { try { TimerTask task; boolean taskFired;
synchronized(queue) { // If the task queue is empty, wait until while (queue.isEmpty() &&
newTasksMayBeScheduled) queue.wait(); if (queue.isEmpty()) break; // Queue is
empty and will forever remain; die // Queue nonempty; look at first evt and do
the right thing long currentTime, executionTime; // Get task from task queue task = queue.
getMin(); synchronized(task.lock) { if (task.state == TimerTask.CANCELLED) {
queue.removeMin(); continue; // No action required, poll queue again }
currentTime= System.currentTimeMillis(); executionTime = task.nextExecutionTime;
if (taskFired = (executionTime<=currentTime)) { if (task.period == 0) { //
Non-repeating, remove queue.removeMin(); task.state = TimerTask.EXECUTED; } else
{ // Repeating task, reschedule queue.rescheduleMin( task.period<0 ? currentTime
- task.period : executionTime + task.period); } } } if (!taskFired) // Task
hasn't yet fired; wait queue.wait(executionTime - currentTime); } //
Perform tasks , We can see that a new thread is not started here , It's blocking execution if (taskFired) // Task fired; run it, holding no
locks task.run(); } catch(InterruptedException e) { } } }
We'll know from here ,Timer
It is a single thread that executes scheduled tasks , In other words, your task may not be executed when it's time , Because the last task has not been completed , Let's write an example here , That's the example above , Simple modification
public static void main(String[] args) throws InterruptedException { timer(); }
// On the whole, we still hope 1s Execute once public static void timer(){ Timer t = new Timer(); t.
scheduleAtFixedRate(new TimerTask() { @Override public void run() { System.out.
println(Thread.currentThread().getName()+ ": hello world"); // We waited in this task
try { TimeUnit.SECONDS.sleep(1000000); } catch (InterruptedException e) { e.
printStackTrace(); } } }, 100, 1000); }
In fact, you will find out after you execute it , It doesn't guarantee your mission TimerTask object 1s Execute once , Because the last task has not been completed . That's why idea
I suggest you use ScheduledExecutorService, It essentially belongs to the category of thread pool , When we study the implementation of thread pool, we will .
<>FutureTask
stay Java Concurrency in program FutureTask Represents an asynchronous operation that can be canceled . It has start and cancel operations , Query whether the operation is completed and retrieve the operation results .
Only when the operation is completed The result can only be retrieved when it is completed , If the operation has not been completed get Method will block .
One FutureTask Objects can be Callable and Runnable Wrap objects , because FutureTask It's also realized Runnable Interface, so it can be submitted to Executor To execute , You can also talk to Thread
Use with , But there's a problem here. That's what we know Runnable There is no return value , that FutureTask How is the return value achieved , Let's have a look
public class FutureTaskDemo { public static void main(String[] args) throws
ExecutionException, InterruptedException { callableDemo(); runableDemo(); }
public static void callableDemo() throws ExecutionException,
InterruptedException { Callable<Integer> call = new Callable<Integer>() {
@Override public Integer call() throws Exception { System.out.println(
" Calculating results ..."); Thread.sleep(3000); return 1; } }; FutureTask<Integer> task = new
FutureTask<>(call); Thread thread = new Thread(task); thread.start(); Integer
result= task.get(); System.out.println("callableDemo The result is ：" + result); } public
static void runableDemo() throws ExecutionException, InterruptedException {
Runnable run = new Runnable() { @SneakyThrows @Override public void run() {
System.out.println(" Calculating results ..."); Thread.sleep(3000); } }; // The return value is predefined by ourselves
FutureTask<Integer> task = new FutureTask(run,1); Thread thread = new Thread(
task); thread.start(); Integer result = task.get(); System.out.println(
"runableDemo The result is ：" + result); } }
In fact, we see FutureTask In essence, it is still used Callable Or Runnable
<> Advanced implementation
<>Java8 lambda expression
This is actually a grammar sugar , It's still an old routine in essence , Let's take a brief look , What the hell is this , It is essentially a functional interface
public class LambdaDemo { public static void main(String[] args) { lambdaThread
(); lambdaRunable(); } public static void lambdaThread() { Thread t = new Thread
(() -> { System.out.println("lambdaThread Implementation of "); }); t.start(); } public
static void lambdaRunable() { Runnable r = () -> { System.out.println(
"lambdaRunable Implementation of "); }; Thread t1 = new Thread(r); Thread t2 = new Thread(()
-> { r.run(); }); t1.start(); t2.start(); } }
<>Java8 stream
This is mainly used Java8 Medium stream api
public class StreamDemo { public static void main(String[] args) {
Stream.of(1,2,3,4,5,6,7,8,9,10).parallel().forEach(ele->{
System.out.println(Thread.currentThread().getName()+":"+ele); }); } }
output : We see multiple threads started
ForkJoinPool.commonPool-worker-1:3 ForkJoinPool.commonPool-worker-1:4
ForkJoinPool.commonPool-worker-5:5 ForkJoinPool.commonPool-worker-5:10
ForkJoinPool.commonPool-worker-4:1 ForkJoinPool.commonPool-worker-2:9
ForkJoinPool.commonPool-worker-3:2 ForkJoinPool.commonPool-worker-5:6
ForkJoinPool.commonPool-worker-1:8 main:7
The default number of threads used by concurrent streams is equal to the number of processor cores on your machine . This method allows you to modify this value , This is a global attribute ,
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism",
"12");
fork/join Frame is jdk1.7 Introduced ,java8 of stream Multithreading is not streaming, which is based on this framework , Therefore, if you want to deeply understand concurrent flow, you need to learn fork/join frame .fork/join The purpose of the framework is to recursively split parallel tasks into smaller tasks , Then combine the results of each subtask to generate the overall results . It is ExecutorService An implementation of interface , It allocates subtasks to the thread pool （ForkJoinPool） Worker threads in . To submit a task to this thread pool , Must create RecursiveTask A subclass of , Yes if the task does not return a result RecursiveAction Subclass of .
<> Thread pool
Here we will not explain in detail what is related to thread pool , Let's talk about how thread pool realizes multi wire , That is, how the thread pool creates threads , We know that the purpose of using thread pool is to avoid manually creating a large number of threads , Give control to the thread pool to achieve the purpose of thread reuse .
First, let's take a look at how we use thread pools
public class ThreadPoolDemo { public static void main(String[] args) throws
InterruptedException { ExecutorService executorService = Executors.
newCachedThreadPool(); while (true) { executorService.submit(() -> { while (true
) { System.out.println(Thread.currentThread().getName()); try { TimeUnit.SECONDS
.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } } });
TimeUnit.SECONDS.sleep(1); } } }
From here we can see , In fact, we only need to submit one to the thread pool Runnable Object can be , In fact, according to our previous passage Runnable We can probably guess how to implement threads , Thread pool is submitted by us Runnable object , Created a thread for us . When we create a thread pool, we actually have such a parameter , Is the factory used to create threads ,
Executors.newCachedThreadPool() newCachedThreadPool The method is just java
A convenient way for us , In fact, the following constructor will be called in the end .
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long
keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(
corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.
defaultThreadFactory(), defaultHandler); }
Now we can have a look at this factory
/** * The default thread factory */ static class DefaultThreadFactory
implements ThreadFactory { private static final AtomicInteger poolNumber = new
AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger
threadNumber= new AtomicInteger(1); private final String namePrefix;
DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group
= (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
namePrefix= "pool-" + poolNumber.getAndIncrement() + "-thread-"; } //
Here is how we create threads public Thread newThread(Runnable r) { Thread t = new Thread(group,
r, namePrefix + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) t.
setDaemon(false); if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(
Thread.NORM_PRIORITY); return t; } }
For line pass pool , Threads are essentially created through a thread factory , Default use DefaultThreadFactory
, It sets some default values for threads created by the thread pool , such as ： Thread name , Is it a daemon thread , And the priority of threads . But no matter how you set these properties , Eventually it passed new
Thread() To create a thread
, But the constructor here passes in more parameters , It can be seen from this that creating threads through thread pool does not deviate from the two basic creation methods at the beginning , Because it's essentially through new Thread()
Realized .
<> Thread startup and status
<>start Methods and run method
This is actually a very old question , That means we have to call start Method will help us start a thread , If you are calling directly run Method words , That's actually a synchronous call .
<> Multiple starts
Let's take a look at what happens after multiple starts
public class ThreadStartTimes { public static void main(String[] args) {
Runnable target; Thread thread = new Thread(()->{ System.out.println(Thread.
currentThread().getName()); }); thread.start(); System.out.println(1); thread.
start(); System.out.println(2); thread.start(); System.out.println(3); } }
The output is as follows :
1 Thread-0 Exception in thread "main" java.lang.IllegalThreadStateException at
java.lang.Thread.start(Thread.java:708) at
thread.thread.ThreadStartTimes.main(ThreadStartTimes.java:12)
We saw that the report was wrong , Let's take a look at the implementation of this method
public synchronized void start() { /** *
In other words, if our thread is in the new state (NEW),threadStatus yes 0, Threads in other states cannot be called start Methodical * A zero status
value corresponds to state "NEW". */ if (threadStatus != 0) throw new
IllegalThreadStateException(); /* Notify the group that this thread is about to
be started * so that it can be added to the group's list of threads * and the
group's unstarted count can be decremented. */ group.add(this); boolean started
= false; try { start0(); started = true; } finally { try { if (!started) { group
.threadStartFailed(this); } } catch (Throwable ignore) { /* do nothing. If
start0 threw a Throwable then it will be passed up the call stack */ } } }
That is, if our thread calls for the first time start
After method threadStatus And No 0 Yes , If you call this method again at this time , Will report an error . But it should be noted that you are Thread
There is no change of state in the code , In other words, the change of state is caused by local Method maintenance
<> Correctly understand threads and Runnable Relationship between
Understanding this is important , This is why thread pools exist , Let's look at the following code , We can see that the core logic of a thread is actually a call target.run(), And here target
namely runnable
@Override public void run() { if (target != null) { target.run(); } }
Because normally our run
Method or our thread object only holds one runnable object , This time often gives people a wrong situation , That's it runnable It is functionally equivalent to thread , In fact that was not the case. , We can look at the following code , A piece of core code of this thread pool , We will also explain in detail later when learning thread pool
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable
task= w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean
completedAbruptly= true; try { // Here's the point while (task != null || (task = getTask()
) != null) { w.lock(); 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); } }
The thread we hold is not a simple one Runnable
object , Then run , When the operation is over , The thread is over , The implementation here is that the thread runs a Runnable object , After running, get another one from the queue Runnable object , That is, a thread runs multiple threads Runnable object , This is the principle of thread pool , This is thread and Runnable Relationship between
<> summary
*
Whether it is Callable still FutureTask, They first and foremost Runnable
equally , It's all a task , It needs to be executed , Not that they are threads themselves . They can be executed in the thread pool , As shown in the code , submit()
Method to put the task into the thread pool , And create threads from the thread pool , In any way , In the end, it is executed by threads , However, the creation of sub threads is still inseparable from the two basic methods mentioned at the beginning , That is to realize
Runnable Interface and inheritance Thread class . Always remember that threads are threads , A task is a task , Thread is Thread The task is Runnable
Something like that , Only in this way can you understand the meaning of thread pool , That is, we give the task to the thread pool , Thread pool uses the threads in the pool to run our tasks , When a task is finished , This thread can run other tasks , So as to achieve the reuse of threads .
*
Callable,FutureTask,Future All in JDK1.5
It was introduced when , So when we look back, there is only one way to implement multithreading , Because other implementations must also be used Thread
Class to implement multithreading , So we can understand other classes like this Callable,FutureTask,Runnable They are all task objects , Then the task object needs to be put into the thread to execute .
*
Thread and Runnable Which is better ,Runnable good
* First, because Java Only single inheritance is allowed , So we can use Runnable To realize the function , Our object can also inherit other required classes
*
The second is us Thread Every creation of is a real thread overhead , So we can actually use one thread object to execute multiple threads Runnable, This is very interesting , This is also the principle of thread pool ;
* Finally, the design of the code , adopt Runnable Implement multithreading , Reached Runnable And Thread Class decoupling ,Thread
Class is responsible for thread startup and property setting ,Runnable
Encapsulate our business logic , That's actually why we can use Thread To execute multiple Runnable object , Because their positioning and original design intention are different ,Thread It's a physical thread , and Runnable Is the business logic to be executed on the thread .
Technology