How to implement message queue in JAVA

Message queue in java

Message queue is a means of communication between threads:

import?java.util.*

public ?class?MsgQueue{

private?Vector?queue?=?null;

public?MsgQueue(){ queue?=?newVector();

}

public?synchronized?void?send(Object?o)

{ queue.addElement(o);

}

public?synchronized?Object?recv()

{ if(queue.size()==0)

return?null; Object?o?=?queue.firstElement() ; queue.removeElementAt(0); //or?queue[0]?=?null?can?also?work return?o;

}

}

Because Java is locked?by?object, adding synchronized? can be used to lock objects synchronously for threads

It can be used as a queue for storing tasks for multi-threading and multi-tasking. Its client includes encapsulated task classes and thread classes

Java's multi-threading - communication between threads 2009-08-25?21:58

1. How many threads are there? States

Threads have four states. Any thread must be in one of these four states:

1)? Generate (New): The thread object has been generated, but It has not been started yet, so it cannot be executed. For example, after a thread object is generated through new but before the start() function is called on it.

2) Executable (Runnable): Every system that supports multi-threading has a scheduler, and the scheduler will select a thread from the thread pool and start it. When a thread is in the executable state, it means that it may be in the thread pool waiting for the scheduler to start it; it may also be executing. For example, after executing the start() method of a thread object, the thread is in an executable state, but it is obvious that the thread is not necessarily executing at this time.

3)?Death: When a thread ends normally, it is in a dead state. For example, after the run() function of a thread is executed, the thread enters the death state.

4) Blocked: When a thread is in a stalled state, the system scheduler will ignore it and not schedule it. When a stalled thread returns to the executable state, it may resume execution. For example, after calling the wait() function on a thread, the thread enters a stagnant state. Only when notify or notifyAll is called on the thread twice can it return to the executable state twice.

2.?Class Commonly used functions under Thread

2.1?suspend(), resume()

1)?Through the suspend() function, you can Put the thread into a stalled state. After a thread is put into a stalled state through suspend(), the thread will not return to the executable state unless a resume() message is received.

2) When calling the suspend() function, the thread will not release its "lock flag".

Example 11:

class?TestThreadMethod?extends?Thread{

public?static?int?shareVar?=?0;

public?TestThreadMethod(String?name){

super(name);

}

public?synchronized?void?run(){

if(shareVar==0){

for(int?i=0;?ilt;5;?i){

shareVar;

if(shareVar==5){

this.suspend(); //(1)

}}}

else{

System.out.print(Thread.currentThread().getName());

System.out.println("?shareVar?=?"? ?shareVar);

this.resume(); //(2)

}}

}

public?class?TestThread{

public?static?void?main(String[]?args){

TestThreadMethod?t1?=?new?TestThreadMethod("t1");

TestThreadMethod?t2?= ?new?TestThreadMethod("t2");

t1.start(); //(5)

//t1.start(); //(3)

t2.start(); //(4)

}}

The operation result is:

t2?shareVar?=?5

i. When the thread generated by t1 of code (5) runs to code (1), the thread enters a stagnant state. Then the scheduler calls up the thread generated by t2 in code (4) from the thread pool. At this time, the shareVar value is not 0, so the statement in else is executed.

ii.? Maybe you will ask, why does t1 not enter the executable state after executing code (2)? As mentioned before, t1 and t2 are threads of two different objects, and codes (1) and (2) only operate on the current object, so the result of the thread generated by t1 executing code (1) is the result of object t1 The current thread enters the stagnant state; and the result of the thread execution code (2) generated by t2 is to return all the stagnant threads in the object t2 to the executable state.

iii.? Now comment out code (4) and remove the comment of code (3), will t1 return to the executable state? The result of the operation is that nothing is output. Why is this happening? Maybe you will think that when the thread generated by code (5) executes code (1), it enters a stagnant state; and the thread generated by code (3) and the thread generated by code (5) belong to the same object. , then when the thread generated by code (3) executes to code (2), the thread generated by code (5) can be executed back to the executable state. But be clear, the suspend() function only puts the current thread into a stagnant state, but does not release the "lock flag" obtained by the current thread. Therefore, when the thread generated by code (5) enters a stagnant state, the thread generated by code (3) still cannot be started because the "lock flag" of the current object is still occupied by the thread generated by code (5).

#p#2.2?sleep()

1)?sleep?() function has a parameter, through which the thread can enter a stagnant state within a specified time. When specified After the time has passed, the thread automatically enters the executable state.

2) When the sleep?() function is called, the thread will not release its "lock flag".

Example 12:

class?TestThreadMethod?extends?Thread{

class?TestThreadMethod?extends?Thread{

public?static ?int?shareVar?=?0;

public?TestThreadMethod(String?name){

super(name);

}

public?synchronized?void?run(){

for(int?i=0;?ilt;3;?i){

System.out.print(Thread .currentThread().getName());

System.out.println("?:?"? ?i);

try{

Thread .sleep(100); //(4)

}

catch(InterruptedException?e){

System.out.println("Interrupted") ;

}}}

}

public?class?TestThread{public?static?void?main(String[]?args){

TestThreadMethod?t1?=?new?TestThreadMethod("t1");

TestThreadMethod?t2?=?new?TestThreadMethod("t2");

t1. start(); (1)

t1.start(); (2)

//t2.start(); (3)

}}

The running result is:

t1?:?0

t1?:?1

t1?:?2

t1?:?0

t1?:?1

t1?:?2

It can be proved by the results that although in run() sleep() is executed, but it will not release the "lock flag" of the object, so unless the thread of code (1) completes the run() function and releases the "lock flag" of the object, the thread of code (2) will never Will not be executed.

If you comment out code (2) and remove the comment from code (3), the result will become:

t1?:?0

t2 ?:?0

t1?:?1

t2?:?1

t1?:?2

t2?: ?2

Since t1 and t2 are threads of two objects, when thread t1 enters stagnation through sleep(), the scheduler will call other executable threads from the thread pool, so that thread t2 is activated.

Example 13:

class?TestThreadMethod?extends?Thread{

public?static?int?shareVar?=?0;

public?TestThreadMethod(String?name){

super(name);

}

public?synchronized?void?run(){

for(int?i=0;?ilt;5;?i){

System.out.print(Thread.currentThread().getName());

System.out.println("?：?"? ?i);

try{

if(Thread.currentThread().getName().equals("t1 "))

Thread.sleep(200);

else

Thread.sleep(100);

}

catch(InterruptedException?e){

System.out.println("Interrupted");

}}

}}

public?class?TestThread{public?static?void?main(String[]?args){

TestThreadMethod?t1?=?new?TestThreadMethod("t1");

TestThreadMethod?t2?=?new?TestThreadMethod("t2");

t1.start();

//t1.start();

t2.start();

}}

The running result is:

t1?:?0

t2 ?:?0

t2?:?1

t1?:?1

t2?:?2

t2?: ?3

t1?:?2

t2?:?4

t1?:?3

t1?:?4

Since thread t1 called sleep(200), and thread t2 called sleep(100), the time that thread t2 was in the stagnant state was half of that of thread t1, and the result reflected in the result is thread t2. Thread t1 prints twice before printing once.

#p#2.3?yield()

1)?Through the yield?() function, the thread can enter the executable state, and the scheduler starts from the thread in the executable state Reschedule. Therefore, the function that calls yield() may also be executed immediately.

2) When the yield?() function is called, the thread will not release its "lock flag".

Example 14:

class?TestThreadMethod?extends?Thread{

public?static?int?shareVar?=?0;

public?TestThreadMethod(String?name){super(name);

}

public?synchronized?void?run(){for(int?i=0;?ilt ;4;?i ){

System.out.print(Thread.currentThread().getName());

System.out.println("?:?"? ?i);

Thread.yield();

}}

}

public?class?TestThread{public?static ?void?main(String[]?args){

TestThreadMethod?t1?=?new?TestThreadMethod("t1");

TestThreadMethod?t2?=?new?TestThreadMethod ("t2");

t1.start();

t1.start(); //(1)

//t2.start( ); (2)

}

}

The running result is:

t1?:?0

t1?：?1

t1?：?2

t1?：?3

t1?：?0

t1 ?:?1

t1?:?2

t1?:?3

It can be seen from the results that the object will not be released when yield() is called. lock sign".

If you comment out code (1) and remove the comment from code (2), the result is:

t1?:?0

t1?: ?1

t2?:?0

t1?:?2

t2?:?1

t1?:?3

t2?:?2

t2?:?3

From the results, we can see that although the t1 thread called yield(), it was executed again immediately. .

2.4 The difference between?sleep() and yield()

1)?sleep() puts the current thread into a stagnant state, so the thread executing sleep() within the specified time It will definitely not be executed; yield() only returns the current thread to the executable state, so the thread executing yield() may be executed again immediately after entering the executable state.

2)?sleep() can give threads with low priority a chance to execute. Of course, it can also give threads with the same priority and high priority a chance to execute; yield() can only give threads with the same priority a chance to execute. Priority threads have a chance to execute.

Example 15:

class?TestThreadMethod?extends?Thread{

public?static?int?shareVar?=?0;

public?TestThreadMethod(String?name){

super(name);

}

public?void?run(){

for(int?i=0;?ilt;4;?i){

System.out.print(Thread.currentThread().getName());

System.out.println("?：?"? ?i);

//Thread.yield(); (1)

/*?(2)?*/

try{

Thread.sleep(3000);

}

catch(InterruptedException?e){

System.out.println("Interrupted");

}}}

}

public?class?TestThread{

public?static?void?main(String[]?args){

TestThreadMethod?t1?=?new?TestThreadMethod("t1");

TestThreadMethod?t2?= ?new?TestThreadMethod("t2");

t1.setPriority(Thread.MAX_PRIORITY);

t2.setPriority(Thread.MIN_PRIORITY);

t1 .start();

t2.start();

}

}

The running result is:

t1?：?0

t1?：?1

t2?：?0

t1?：?2

t2?:?1

t1?:?3

t2?:?2

t2?:?3

By the result It can be seen that sleep() can give threads with lower priority a chance to execute. Comment out code (2) and remove the comment of code (1). The result is:

t1?:?0

t1?:?1

t1?：?2

t1?：?3

t2?：?0

t2?：?1

t2? :?2

t2?:?3

It can be seen that when yield() is called, threads with different priorities will never get the opportunity to execute.

2.5?join()

The thread calling join() can only execute other threads after the thread calling join() is completed. In a certain sense, it can realize the synchronization function.

Example 16:

class?TestThreadMethod?extends?Thread{

public?static?int?shareVar?=?0;

public?TestThreadMethod(String?name){

super(name);

}

public?void?run(){

for(int?i=0;?ilt;4;?i){

System.out.println("?"? ?i);

try{

Thread.sleep(3000);

}

catch(InterruptedException?e){

System.out.println("Interrupted ");

}

}

}

}

public?class?TestThread{

public?static?void?main(String[]?args){

TestThreadMethod?t1?=?new?TestThreadMethod("t1");

t1 .start();

try{

t1.join();

}

catch(InterruptedException?e){}

t1.start();

}

}

The running result is:

1

2

3

1

2

3

#p#3. ?class | Commonly used thread functions under Object

The three functions wait(), notify() and notifyAll() are provided by the java.lang.Object class and are used to coordinate multiple threads to share Data access.

3.1?wait(), notify() and notifyAll()

1) The wait() function has two forms: the first form accepts a millisecond value for Pauses a thread for a specified length of time, causing the thread to enter a stalled state. The second form takes no parameters, which means wait() will continue to stall before notify() or notifyAll().

2)? When notify() is executed on an object, any thread will be removed from the thread waiting pool and placed in the lock flag waiting pool; when notifyAll is executed on an object (), all threads of this object will be removed from the thread waiting pool and placed in the lock flag waiting pool.

3) When wait() is called, the thread will release the "lock flag" it occupies, so that other synchronized data in the object where the thread is located can be used by other threads.

Example 17:

Next, we will modify the example in Example 11

class?TestThreadMethod?extends?Thread{

public?static?int?shareVar?=?0;

public?TestThreadMethod(String?name){

super(name);

}

public?synchronized?void?run(){

if(shareVar==0){

for(int?i=0;?ilt;10 ;?i ){

shareVar ;

if(shareVar==5){

try{

this.wait() ; //(4)

}

catch(InterruptedException?e){}

}

}

}

if(shareVar!=0){

System.out.print(Thread.currentThread().getName());

System .out.println("?shareVar?=?"? ?shareVar);

this.notify(); //(5)

}

}

}

public?class?TestThread{

public?static?void?main(String[]?args){

TestThreadMethod?t1?=?new?TestThreadMethod("t1");

TestThreadMethod?t2?=?new?TestThreadMethod("t2");

t1.start() ; //(1)

//t1.start(); (2)

t2.start(); //(3)

} }

The running result is:

t2?shareVar?=?5

Because t1 and t2 are two different objects, thread t2 calls the code (5 ) cannot wake up thread t1. If you remove the comment of code (2) and comment out code (3), the result is:

t1?shareVar?=?5

t1?shareVar?=?10

This is because when the thread of code (1) executes to code (4), it enters a stalled state and releases the object's lock state. Then, the thread of code (2) executes run(). Since the shareVar value is 5 at this time, the print statement is executed and code (5) is called to make the thread of code (1) enter the executable state, and then the thread of code (2) Finish. When the thread of code (1) is re-executed, it then executes the for() loop until shareVar=10, and then prints shareVar.

#p#3.2?wait(), notify() and synchronized

Waite() and notify() operate on the "lock flag" of the object, so they must Called in synchronized function or synchronized block.

If the call is made in a non-synchronized function or non-synchronized block, although it can be compiled, an IllegalMonitorStateException exception will occur at runtime.

Example 18:

class?TestThreadMethod?extends?Thread{

public?int?shareVar?=?0;

public ?TestThreadMethod(String?name){

super(name);

new?Notifier(this);

}

public ?synchronized?void?run(){

if(shareVar==0){

for(int?i=0;?ilt;5;?i){

shareVar;

System.out.println("i?=?"? ?shareVar);

try{

System.out .println("wait...");

this.wait();

}

catch(InterruptedException?e){}

}}

}

}

class?Notifier?extends?Thread{

private?TestThreadMethod ?ttm;

Notifier(TestThreadMethod?t){

ttm?=?t;

start();

}

public?void?run(){

while(true){

try{

sleep(2000);

}

catch(InterruptedException?e){}

/*1? How to synchronize something other than the current object?*/

synchronized (ttm){

System.out.println("notify...");

ttm.notify();

}}

}

}

public?class?TestThread{

public?static?void?main(String[]?args) {

TestThreadMethod?t1?=?new?TestThreadMethod("t1");

t1.start();

}

}

The running result is:

i?=?1

wait...

notify... ...

i?=?2

wait...

notify...

i?=?3

wait......

notify......

i?=?4

wait......

notify......

i?=?5

wait......

notify......

4.? Discussion of wait(), notify(), notifyAll() and suspend(), resume(), sleep()

4.1? The difference between these two sets of functions

1) When ?wait() causes the current thread to enter a stagnant state, it will also release the "lock flag" occupied by the current thread, so that the synchronized resources in the thread object can be used by other threads in the object; while suspend() and sleep () When the current thread enters a stagnant state, the "lock flag" occupied by the current thread will not be released.

2) The former group of functions must be called in synchronized functions or synchronized blocks, otherwise errors will occur at runtime; while the latter group of functions can be called in non-synchronized functions and synchronized blocks.

4.2? The trade-offs between these two groups of functions

Java2 no longer recommends using the latter group of functions. Because the "lock flag" obtained by the current thread will not be released when suspend() is called, this can easily cause a "deadlock".